PamGram/core/dependence/assets.py

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

439 lines
17 KiB
Python
Raw Permalink Normal View History

import asyncio
from pathlib import Path
from ssl import SSLZeroReturnError
2023-04-27 12:25:06 +00:00
from typing import Optional, List, Dict
from aiofiles import open as async_open
2023-04-27 12:25:06 +00:00
from httpx import AsyncClient, HTTPError
from core.base_service import BaseService
2023-04-27 12:25:06 +00:00
from modules.wiki.base import WikiModel
from modules.wiki.models.avatar_config import AvatarIcon
from modules.wiki.models.head_icon import HeadIcon
2023-04-27 12:25:06 +00:00
from modules.wiki.models.light_cone_config import LightConeIcon
from modules.wiki.models.phone_theme import PhoneTheme
2023-04-27 12:25:06 +00:00
from utils.const import PROJECT_ROOT
from utils.log import logger
2023-04-27 12:25:06 +00:00
from utils.typedefs import StrOrURL, StrOrInt
2022-10-10 03:37:58 +00:00
ASSETS_PATH = PROJECT_ROOT.joinpath("resources/assets")
ASSETS_PATH.mkdir(exist_ok=True, parents=True)
2023-04-27 12:25:06 +00:00
DATA_MAP = {
"avatar": WikiModel.BASE_URL + "avatar_icons.json",
"light_cone": WikiModel.BASE_URL + "light_cone_icons.json",
2023-05-11 15:09:39 +00:00
"avatar_eidolon": WikiModel.BASE_URL + "avatar_eidolon_icons.json",
"avatar_skill": WikiModel.BASE_URL + "skill/info.json",
"head_icon": WikiModel.BASE_URL + "head_icons.json",
"phone_theme": WikiModel.BASE_URL + "phone_themes.json",
2023-04-27 12:25:06 +00:00
}
class AssetsServiceError(Exception):
pass
class AssetsCouldNotFound(AssetsServiceError):
def __init__(self, message: str, target: str):
self.message = message
self.target = target
2024-06-05 11:23:18 +00:00
super().__init__(f"{message}: target={target}")
2023-04-27 12:25:06 +00:00
class _AssetsService:
client: Optional[AsyncClient] = None
def __init__(self, client: Optional[AsyncClient] = None) -> None:
2023-04-27 12:25:06 +00:00
self.client = client
2023-04-27 12:25:06 +00:00
async def _download(self, url: StrOrURL, path: Path, retry: int = 5) -> Optional[Path]:
"""从 url 下载图标至 path"""
if not url:
return None
logger.debug("正在从 %s 下载图标至 %s", url, path)
2023-03-31 06:01:03 +00:00
headers = None
for time in range(retry):
try:
response = await self.client.get(url, follow_redirects=False, headers=headers)
except Exception as error: # pylint: disable=W0703
if not isinstance(error, (HTTPError, SSLZeroReturnError)):
logger.error(error) # 打印未知错误
if time != retry - 1: # 未达到重试次数
await asyncio.sleep(1)
else:
raise error
continue
if response.status_code != 200: # 判定页面是否正常
return None
2022-10-10 03:37:58 +00:00
async with async_open(path, "wb") as file:
await file.write(response.content) # 保存图标
return path.resolve()
2023-04-27 12:25:06 +00:00
class _AvatarAssets(_AssetsService):
path: Path
data: List[AvatarIcon]
name_map: Dict[str, AvatarIcon]
id_map: Dict[int, AvatarIcon]
def __init__(self, client: Optional[AsyncClient] = None) -> None:
super().__init__(client)
self.path = ASSETS_PATH.joinpath("avatar")
self.path.mkdir(exist_ok=True, parents=True)
async def initialize(self):
logger.info("正在初始化角色素材图标")
html = await self.client.get(DATA_MAP["avatar"])
2023-05-11 15:09:39 +00:00
eidolons = await self.client.get(DATA_MAP["avatar_eidolon"])
eidolons_data = eidolons.json()
skills = await self.client.get(DATA_MAP["avatar_skill"])
skills_data = skills.json()
2023-04-27 12:25:06 +00:00
self.data = [AvatarIcon(**data) for data in html.json()]
self.name_map = {icon.name: icon for icon in self.data}
self.id_map = {icon.id: icon for icon in self.data}
tasks = []
for icon in self.data:
2023-05-11 15:09:39 +00:00
eidolons_s_data = eidolons_data.get(str(icon.id), [])
skills_s_data = [f"{i}.png" for i in skills_data if i.startswith(str(icon.id) + "_")]
2023-04-27 12:25:06 +00:00
base_path = self.path / f"{icon.id}"
base_path.mkdir(exist_ok=True, parents=True)
gacha_path = base_path / "gacha.webp"
icon_path = base_path / "icon.webp"
normal_path = base_path / "normal.webp"
2023-04-28 12:03:38 +00:00
square_path = base_path / "square.png"
2023-05-11 15:09:39 +00:00
eidolons_paths = [(base_path / f"eidolon_{eidolon_id}.webp") for eidolon_id in range(1, 7)]
skills_paths = []
for i in skills_s_data:
temp_end = "_".join(i.split("_")[1:])
skills_paths.append(base_path / f"skill_{temp_end}")
2023-04-27 12:25:06 +00:00
if not gacha_path.exists():
tasks.append(self._download(icon.gacha, gacha_path))
if not icon_path.exists():
tasks.append(self._download(icon.icon_, icon_path))
if not normal_path.exists():
tasks.append(self._download(icon.normal, normal_path))
2023-04-28 12:03:38 +00:00
if not square_path.exists() and icon.square:
tasks.append(self._download(icon.square, square_path))
2023-05-11 15:09:39 +00:00
for index, eidolon in enumerate(eidolons_paths):
2023-06-06 15:27:43 +00:00
if not eidolons_s_data:
break
2023-05-11 15:09:39 +00:00
if not eidolon.exists():
tasks.append(self._download(eidolons_s_data[index], eidolon))
for index, skill in enumerate(skills_paths):
if not skill.exists():
tasks.append(self._download(WikiModel.BASE_URL + "skill/" + skills_s_data[index], skill))
2023-04-27 12:25:06 +00:00
if len(tasks) >= 100:
await asyncio.gather(*tasks)
tasks = []
if tasks:
await asyncio.gather(*tasks)
logger.info("角色素材图标初始化完成")
2023-04-28 12:03:38 +00:00
def get_path(self, icon: AvatarIcon, name: str, ext: str = "webp") -> Path:
2023-04-27 12:25:06 +00:00
path = self.path / f"{icon.id}"
path.mkdir(exist_ok=True, parents=True)
2023-04-28 12:03:38 +00:00
return path / f"{name}.{ext}"
2023-04-27 12:25:06 +00:00
def get_by_id(self, id_: int) -> Optional[AvatarIcon]:
return self.id_map.get(id_, None)
def get_by_name(self, name: str) -> Optional[AvatarIcon]:
return self.name_map.get(name, None)
2023-04-29 16:15:21 +00:00
def get_target(self, target: StrOrInt, second_target: StrOrInt = None) -> AvatarIcon:
2023-04-28 12:03:38 +00:00
data = None
2023-04-27 12:25:06 +00:00
if isinstance(target, int):
2023-04-28 12:03:38 +00:00
data = self.get_by_id(target)
2023-04-27 12:25:06 +00:00
elif isinstance(target, str):
2023-04-28 12:03:38 +00:00
data = self.get_by_name(target)
if data is None:
2023-04-29 16:15:21 +00:00
if second_target:
return self.get_target(second_target)
2023-04-28 12:03:38 +00:00
raise AssetsCouldNotFound("角色素材图标不存在", target)
return data
2023-04-29 16:15:21 +00:00
def gacha(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
2023-04-27 12:25:06 +00:00
return self.get_path(icon, "gacha")
2023-04-29 16:15:21 +00:00
def icon(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
2023-04-27 12:25:06 +00:00
return self.get_path(icon, "icon")
2023-04-29 16:15:21 +00:00
def normal(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
2023-04-27 12:25:06 +00:00
return self.get_path(icon, "normal")
2023-04-29 16:15:21 +00:00
def square(self, target: StrOrInt, second_target: StrOrInt = None, allow_icon: bool = True) -> Path:
icon = self.get_target(target, second_target)
2023-04-28 12:03:38 +00:00
path = self.get_path(icon, "square", "png")
if not path.exists():
if allow_icon:
return self.get_path(icon, "icon")
raise AssetsCouldNotFound("角色素材图标不存在", target)
return path
2023-05-11 15:09:39 +00:00
def eidolons(self, target: StrOrInt, second_target: StrOrInt = None) -> List[Path]:
"""星魂"""
icon = self.get_target(target, second_target)
return [self.get_path(icon, f"eidolon_{i}") for i in range(1, 7)]
def skill_basic_atk(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
"""普攻 001"""
icon = self.get_target(target, second_target)
return self.get_path(icon, "skill_basic_atk", "png")
def skill_skill(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
"""战技 002"""
icon = self.get_target(target, second_target)
return self.get_path(icon, "skill_skill", "png")
def skill_ultimate(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
"""终结技 003"""
icon = self.get_target(target, second_target)
return self.get_path(icon, "skill_ultimate", "png")
def skill_talent(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
"""天赋 004"""
icon = self.get_target(target, second_target)
return self.get_path(icon, "skill_talent", "png")
def skill_technique(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
"""秘技 007"""
icon = self.get_target(target, second_target)
return self.get_path(icon, "skill_technique", "png")
def skills(self, target: StrOrInt, second_target: StrOrInt = None) -> List[Path]:
icon = self.get_target(target, second_target)
return [
self.get_path(icon, "skill_basic_atk", "png"),
self.get_path(icon, "skill_skill", "png"),
self.get_path(icon, "skill_ultimate", "png"),
self.get_path(icon, "skill_talent", "png"),
self.get_path(icon, "skill_technique", "png"),
]
2023-04-27 12:25:06 +00:00
class _LightConeAssets(_AssetsService):
path: Path
data: List[LightConeIcon]
name_map: Dict[str, LightConeIcon]
id_map: Dict[int, LightConeIcon]
def __init__(self, client: Optional[AsyncClient] = None) -> None:
super().__init__(client)
2023-04-27 12:25:06 +00:00
self.path = ASSETS_PATH.joinpath("light_cone")
self.path.mkdir(exist_ok=True, parents=True)
async def initialize(self):
logger.info("正在初始化光锥素材图标")
html = await self.client.get(DATA_MAP["light_cone"])
self.data = [LightConeIcon(**data) for data in html.json()]
self.name_map = {icon.name: icon for icon in self.data}
self.id_map = {icon.id: icon for icon in self.data}
tasks = []
for icon in self.data:
base_path = self.path / f"{icon.id}"
base_path.mkdir(exist_ok=True, parents=True)
gacha_path = base_path / "gacha.webp"
icon_path = base_path / "icon.webp"
if not gacha_path.exists():
tasks.append(self._download(icon.gacha, gacha_path))
if not icon_path.exists():
tasks.append(self._download(icon.icon_, icon_path))
if len(tasks) >= 100:
await asyncio.gather(*tasks)
tasks = []
if tasks:
await asyncio.gather(*tasks)
logger.info("光锥素材图标初始化完成")
def get_path(self, icon: LightConeIcon, name: str) -> Path:
path = self.path / f"{icon.id}"
path.mkdir(exist_ok=True, parents=True)
return path / f"{name}.webp"
def get_by_id(self, id_: int) -> Optional[LightConeIcon]:
return self.id_map.get(id_, None)
def get_by_name(self, name: str) -> Optional[LightConeIcon]:
return self.name_map.get(name, None)
2023-04-29 16:15:21 +00:00
def get_target(self, target: StrOrInt, second_target: StrOrInt = None) -> Optional[LightConeIcon]:
2023-04-27 12:25:06 +00:00
if isinstance(target, int):
return self.get_by_id(target)
elif isinstance(target, str):
return self.get_by_name(target)
2023-04-29 16:15:21 +00:00
if second_target:
return self.get_target(second_target)
raise AssetsCouldNotFound("光锥素材图标不存在", target)
2023-04-29 16:15:21 +00:00
def gacha(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
2023-04-27 12:25:06 +00:00
return self.get_path(icon, "gacha")
2023-04-29 16:15:21 +00:00
def icon(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
2023-04-27 12:25:06 +00:00
return self.get_path(icon, "icon")
class _HeadIconAssets(_AssetsService):
path: Path
data: List[HeadIcon]
id_map: Dict[int, HeadIcon]
avatar_id_map: Dict[int, HeadIcon]
def __init__(self, client: Optional[AsyncClient] = None) -> None:
super().__init__(client)
self.path = ASSETS_PATH.joinpath("head_icon")
self.path.mkdir(exist_ok=True, parents=True)
async def initialize(self):
logger.info("正在初始化头像素材图标")
html = await self.client.get(DATA_MAP["head_icon"])
self.data = [HeadIcon(**data) for data in html.json()]
self.id_map = {icon.id: icon for icon in self.data}
self.avatar_id_map = {icon.avatar_id: icon for icon in self.data if icon.avatar_id}
tasks = []
for icon in self.data:
webp_path = self.path / f"{icon.id}.webp"
png_path = self.path / f"{icon.id}.png"
if not webp_path.exists() and icon.webp:
tasks.append(self._download(icon.webp, webp_path))
if not png_path.exists() and icon.png:
tasks.append(self._download(icon.png, png_path))
if len(tasks) >= 100:
await asyncio.gather(*tasks)
tasks = []
if tasks:
await asyncio.gather(*tasks)
logger.info("头像素材图标初始化完成")
def get_path(self, icon: HeadIcon, ext: str) -> Path:
path = self.path / f"{icon.id}.{ext}"
return path
def get_by_id(self, id_: int) -> Optional[HeadIcon]:
return self.id_map.get(id_, None)
def get_by_avatar_id(self, avatar_id: int) -> Optional[HeadIcon]:
return self.avatar_id_map.get(avatar_id, None)
def get_target(self, target: StrOrInt, second_target: StrOrInt = None) -> Optional[HeadIcon]:
if target and 1000 < target <= 9000:
data = self.get_by_avatar_id(target)
if data:
return data
data = self.get_by_id(target)
if data:
return data
if second_target:
return self.get_target(second_target)
raise AssetsCouldNotFound("头像素材图标不存在", target)
def webp(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
return self.get_path(icon, "webp")
def png(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
return self.get_path(icon, "png")
def icon(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
icon = self.get_target(target, second_target)
webp_path = self.get_path(icon, "webp")
png_path = self.get_path(icon, "png")
if webp_path.exists():
return webp_path
if png_path.exists():
return png_path
raise AssetsCouldNotFound("头像素材图标不存在", target)
class _PhoneThemeAssets(_AssetsService):
path: Path
data: List[PhoneTheme]
id_map: Dict[int, PhoneTheme]
def __init__(self, client: Optional[AsyncClient] = None) -> None:
super().__init__(client)
self.path = ASSETS_PATH.joinpath("phone_theme")
self.path.mkdir(exist_ok=True, parents=True)
async def initialize(self):
logger.info("正在初始化手机壁纸素材图标")
html = await self.client.get(DATA_MAP["phone_theme"])
self.data = [PhoneTheme(**data) for data in html.json()]
self.id_map = {theme.id: theme for theme in self.data}
tasks = []
for theme in self.data:
path = self.path / f"{theme.id}.png"
if not path.exists():
if theme.urls[0]:
tasks.append(self._download(theme.urls[0], path))
elif theme.urls[1]:
tasks.append(self._download(theme.urls[1], path))
if len(tasks) >= 100:
await asyncio.gather(*tasks)
tasks = []
if tasks:
await asyncio.gather(*tasks)
logger.info("手机壁纸素材图标初始化完成")
def get_path(self, theme: PhoneTheme, ext: str) -> Path:
path = self.path / f"{theme.id}.{ext}"
return path
def get_by_id(self, id_: int) -> Optional[PhoneTheme]:
return self.id_map.get(id_, None)
def get_target(self, target: StrOrInt, second_target: StrOrInt = None) -> Optional[PhoneTheme]:
data = self.get_by_id(target)
if data:
return data
if second_target:
return self.get_target(second_target)
raise AssetsCouldNotFound("手机壁纸素材图标不存在", target)
def icon(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
theme = self.get_target(target, second_target)
png_path = self.get_path(theme, "png")
if png_path.exists():
return png_path
raise AssetsCouldNotFound("手机壁纸素材图标不存在", target)
class AssetsService(BaseService.Dependence):
"""asset服务
用于储存和管理 asset :
当对应的 asset (如某角色图标)不存在时该服务会先查找本地
若本地不存在则从网络上下载若存在则返回其路径
"""
2023-04-27 12:25:06 +00:00
client: Optional[AsyncClient] = None
avatar: _AvatarAssets
"""角色"""
head_icon: _HeadIconAssets
"""头像"""
phone_theme: _PhoneThemeAssets
"""手机壁纸"""
2023-04-27 12:25:06 +00:00
light_cone: _LightConeAssets
"""光锥"""
def __init__(self):
2023-04-27 12:25:06 +00:00
self.client = AsyncClient(timeout=60.0)
self.avatar = _AvatarAssets(self.client)
self.head_icon = _HeadIconAssets(self.client)
self.phone_theme = _PhoneThemeAssets(self.client)
2023-04-27 12:25:06 +00:00
self.light_cone = _LightConeAssets(self.client)
2023-04-27 12:25:06 +00:00
async def initialize(self): # pylint: disable=W0221
await self.avatar.initialize()
await self.head_icon.initialize()
await self.phone_theme.initialize()
2023-04-27 12:25:06 +00:00
await self.light_cone.initialize()