mirror of
https://github.com/PaiGramTeam/PaiGram.git
synced 2024-11-21 22:58:05 +00:00
🐛 Fix AssetsService
This commit is contained in:
parent
d4f1d64daf
commit
94ba52e7fc
@ -6,7 +6,15 @@ from functools import cached_property, lru_cache, partial
|
|||||||
from multiprocessing import RLock as Lock
|
from multiprocessing import RLock as Lock
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from ssl import SSLZeroReturnError
|
from ssl import SSLZeroReturnError
|
||||||
from typing import AsyncIterator, Awaitable, Callable, ClassVar, Dict, Optional, TYPE_CHECKING, TypeVar, Union
|
from typing import (
|
||||||
|
AsyncIterator,
|
||||||
|
Awaitable,
|
||||||
|
Callable,
|
||||||
|
ClassVar,
|
||||||
|
TYPE_CHECKING,
|
||||||
|
TypeVar,
|
||||||
|
Union,
|
||||||
|
)
|
||||||
|
|
||||||
from aiofiles import open as async_open
|
from aiofiles import open as async_open
|
||||||
from aiofiles.os import remove as async_remove
|
from aiofiles.os import remove as async_remove
|
||||||
@ -35,8 +43,8 @@ if TYPE_CHECKING:
|
|||||||
|
|
||||||
__all__ = ("AssetsServiceType", "AssetsService", "AssetsServiceError", "AssetsCouldNotFound", "DEFAULT_EnkaAssets")
|
__all__ = ("AssetsServiceType", "AssetsService", "AssetsServiceError", "AssetsCouldNotFound", "DEFAULT_EnkaAssets")
|
||||||
|
|
||||||
ICON_TYPE = Union[Callable[[bool], Awaitable[Optional[Path]]], Callable[..., Awaitable[Optional[Path]]]]
|
ICON_TYPE = Union[Callable[[bool], Awaitable[Path | None]], Callable[..., Awaitable[Path | None]]]
|
||||||
NAME_MAP_TYPE = Dict[str, StrOrURL]
|
NAME_MAP_TYPE = dict[str, StrOrURL]
|
||||||
|
|
||||||
ASSETS_PATH = PROJECT_ROOT.joinpath("resources/assets")
|
ASSETS_PATH = PROJECT_ROOT.joinpath("resources/assets")
|
||||||
ASSETS_PATH.mkdir(exist_ok=True, parents=True)
|
ASSETS_PATH.mkdir(exist_ok=True, parents=True)
|
||||||
@ -62,7 +70,7 @@ class _AssetsService(ABC):
|
|||||||
_dir: ClassVar[Path]
|
_dir: ClassVar[Path]
|
||||||
icon_types: ClassVar[list[str]]
|
icon_types: ClassVar[list[str]]
|
||||||
|
|
||||||
_client: Optional[AsyncClient] = None
|
_client: AsyncClient | None = None
|
||||||
_links: dict[str, str] = {}
|
_links: dict[str, str] = {}
|
||||||
|
|
||||||
id: int
|
id: int
|
||||||
@ -90,7 +98,7 @@ class _AssetsService(ABC):
|
|||||||
self._client = AsyncClient()
|
self._client = AsyncClient()
|
||||||
return self._client
|
return self._client
|
||||||
|
|
||||||
def __init__(self, client: Optional[AsyncClient] = None) -> None:
|
def __init__(self, client: AsyncClient | None = None) -> None:
|
||||||
self._client = client
|
self._client = client
|
||||||
|
|
||||||
def __call__(self, target: int) -> Self:
|
def __call__(self, target: int) -> Self:
|
||||||
@ -239,12 +247,12 @@ class _AvatarAssets(_AssetsService):
|
|||||||
return re.findall(r"UI_AvatarIcon_(.*)", icon)[0]
|
return re.findall(r"UI_AvatarIcon_(.*)", icon)[0]
|
||||||
|
|
||||||
@cached_property
|
@cached_property
|
||||||
def enka(self) -> Optional[EnkaCharacterAsset]:
|
def enka(self) -> EnkaCharacterAsset | None:
|
||||||
api = getattr(self, "_enka_api", None)
|
api = getattr(self, "_enka_api", None)
|
||||||
cid = getattr(self, "id", None)
|
cid = getattr(self, "id", None)
|
||||||
return None if api is None or cid is None else api.character(cid)
|
return None if api is None or cid is None else api.character(cid)
|
||||||
|
|
||||||
def __init__(self, client: Optional[AsyncClient] = None, enka: Optional[EnkaAssets] = None):
|
def __init__(self, client: AsyncClient | None = None, enka: EnkaAssets | None = None):
|
||||||
super().__init__(client)
|
super().__init__(client)
|
||||||
self._enka_api = enka or DEFAULT_EnkaAssets
|
self._enka_api = enka or DEFAULT_EnkaAssets
|
||||||
|
|
||||||
@ -462,11 +470,16 @@ class AssetsService(BaseService.Dependence):
|
|||||||
namecard: _NamecardAssets
|
namecard: _NamecardAssets
|
||||||
"""名片"""
|
"""名片"""
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self, client: AsyncClient | None = None) -> None:
|
||||||
for attr, assets_type_name in filter(
|
for assets_type_name, assets_type in filter(
|
||||||
lambda x: (not x[0].startswith("_")) and x[1].endswith("Assets"), self.__annotations__.items()
|
lambda x: (not x[0].startswith("_")) and x[1].__name__.endswith("Assets"), self.__annotations__.items()
|
||||||
):
|
):
|
||||||
setattr(self, attr, globals()[assets_type_name]())
|
setattr(self, assets_type_name, globals()[assets_type.__name__](client))
|
||||||
|
|
||||||
|
def __iter__(self):
|
||||||
|
for assets_type_name in self.__annotations__.keys():
|
||||||
|
yield getattr(self, assets_type_name)
|
||||||
|
return None
|
||||||
|
|
||||||
async def initialize(self) -> None: # pylint: disable=R0201
|
async def initialize(self) -> None: # pylint: disable=R0201
|
||||||
"""启动 AssetsService 服务,刷新元数据"""
|
"""启动 AssetsService 服务,刷新元数据"""
|
||||||
|
@ -3,7 +3,7 @@ from __future__ import annotations
|
|||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
from functools import partial
|
from functools import partial
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Awaitable, Callable, ClassVar, TypeVar
|
from typing import Awaitable, Callable, ClassVar, Generator, TypeVar
|
||||||
|
|
||||||
from enkanetwork import Assets as EnkaAssets
|
from enkanetwork import Assets as EnkaAssets
|
||||||
from enkanetwork.model.assets import CharacterAsset as EnkaCharacterAsset
|
from enkanetwork.model.assets import CharacterAsset as EnkaCharacterAsset
|
||||||
@ -152,4 +152,6 @@ class AssetsService(BaseService.Dependence):
|
|||||||
namecard: _NamecardAssets
|
namecard: _NamecardAssets
|
||||||
"""名片"""
|
"""名片"""
|
||||||
|
|
||||||
|
def __iter__(self) -> Generator[_AssetsService, None, None]: ...
|
||||||
|
|
||||||
AssetsServiceType = TypeVar("AssetsServiceType", bound=_AssetsService)
|
AssetsServiceType = TypeVar("AssetsServiceType", bound=_AssetsService)
|
||||||
|
Loading…
Reference in New Issue
Block a user