PamGram/core/dependence/assets.py

import asyncio
from pathlib import Path
from ssl import SSLZeroReturnError
from typing import Optional, List, Dict

from aiofiles import open as async_open
from httpx import AsyncClient, HTTPError

from core.base_service import BaseService
from modules.wiki.base import WikiModel
from modules.wiki.models.avatar_config import AvatarIcon
from modules.wiki.models.head_icon import HeadIcon
from modules.wiki.models.light_cone_config import LightConeIcon
from modules.wiki.models.phone_theme import PhoneTheme
from utils.const import PROJECT_ROOT
from utils.log import logger
from utils.typedefs import StrOrURL, StrOrInt

ASSETS_PATH = PROJECT_ROOT.joinpath("resources/assets")
ASSETS_PATH.mkdir(exist_ok=True, parents=True)
DATA_MAP = {
    "avatar": WikiModel.BASE_URL + "avatar_icons.json",
    "light_cone": WikiModel.BASE_URL + "light_cone_icons.json",
    "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",
}


class AssetsServiceError(Exception):
    pass


class AssetsCouldNotFound(AssetsServiceError):
    def __init__(self, message: str, target: str):
        self.message = message
        self.target = target
        super().__init__(f"{message}: target={message}")


class _AssetsService:
    client: Optional[AsyncClient] = None

    def __init__(self, client: Optional[AsyncClient] = None) -> None:
        self.client = client

    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)
        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
            async with async_open(path, "wb") as file:
                await file.write(response.content)  # 保存图标
            return path.resolve()


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"])
        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()
        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:
            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) + "_")]
            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"
            square_path = base_path / "square.png"
            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}")
            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))
            if not square_path.exists() and icon.square:
                tasks.append(self._download(icon.square, square_path))
            for index, eidolon in enumerate(eidolons_paths):
                if not eidolons_s_data:
                    break
                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))
            if len(tasks) >= 100:
                await asyncio.gather(*tasks)
                tasks = []
        if tasks:
            await asyncio.gather(*tasks)
        logger.info("角色素材图标初始化完成")

    def get_path(self, icon: AvatarIcon, name: str, ext: str = "webp") -> Path:
        path = self.path / f"{icon.id}"
        path.mkdir(exist_ok=True, parents=True)
        return path / f"{name}.{ext}"

    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)

    def get_target(self, target: StrOrInt, second_target: StrOrInt = None) -> AvatarIcon:
        data = None
        if isinstance(target, int):
            data = self.get_by_id(target)
        elif isinstance(target, str):
            data = self.get_by_name(target)
        if data is None:
            if second_target:
                return self.get_target(second_target)
            raise AssetsCouldNotFound("角色素材图标不存在", target)
        return data

    def gacha(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
        icon = self.get_target(target, second_target)
        return self.get_path(icon, "gacha")

    def icon(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
        icon = self.get_target(target, second_target)
        return self.get_path(icon, "icon")

    def normal(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
        icon = self.get_target(target, second_target)
        return self.get_path(icon, "normal")

    def square(self, target: StrOrInt, second_target: StrOrInt = None, allow_icon: bool = True) -> Path:
        icon = self.get_target(target, second_target)
        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

    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"),
        ]


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

    def get_target(self, target: StrOrInt, second_target: StrOrInt = None) -> Optional[LightConeIcon]:
        if isinstance(target, int):
            return self.get_by_id(target)
        elif isinstance(target, str):
            return self.get_by_name(target)
        if second_target:
            return self.get_target(second_target)
        raise AssetsCouldNotFound("光锥素材图标不存在", target)

    def gacha(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
        icon = self.get_target(target, second_target)
        return self.get_path(icon, "gacha")

    def icon(self, target: StrOrInt, second_target: StrOrInt = None) -> Path:
        icon = self.get_target(target, second_target)
        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 (如某角色图标)不存在时，该服务会先查找本地。
        若本地不存在，则从网络上下载；若存在，则返回其路径
    """

    client: Optional[AsyncClient] = None

    avatar: _AvatarAssets
    """角色"""

    head_icon: _HeadIconAssets
    """头像"""

    phone_theme: _PhoneThemeAssets
    """手机壁纸"""

    light_cone: _LightConeAssets
    """光锥"""

    def __init__(self):
        self.client = AsyncClient(timeout=60.0)
        self.avatar = _AvatarAssets(self.client)
        self.head_icon = _HeadIconAssets(self.client)
        self.phone_theme = _PhoneThemeAssets(self.client)
        self.light_cone = _LightConeAssets(self.client)

    async def initialize(self):  # pylint: disable=W0221
        await self.avatar.initialize()
        await self.head_icon.initialize()
        await self.phone_theme.initialize()
        await self.light_cone.initialize()