✨ add method get_link for AssetsService

core/base/assets.py

@@ -4,17 +4,18 @@ from __future__ import annotations
 import asyncio
 import re
 from abc import ABC, abstractmethod
+from asyncio import Lock as AsyncLock
 from functools import cached_property, partial
 from multiprocessing import RLock as Lock
 from pathlib import Path
 from ssl import SSLZeroReturnError
-from typing import Awaitable, Callable, ClassVar, Dict, Optional, TYPE_CHECKING, TypeVar, Union
+from typing import AsyncIterator, Awaitable, Callable, ClassVar, Dict, Optional, TYPE_CHECKING, TypeVar, Union
 
 from aiofiles import open as async_open
 from aiofiles.os import remove as async_remove
 from enkanetwork import Assets as EnkaAssets
 from enkanetwork.model.assets import CharacterAsset as EnkaCharacterAsset
-from httpx import AsyncClient, HTTPError, URL
+from httpx import AsyncClient, HTTPError, HTTPStatusError, URL
 from typing_extensions import Self
 
 from core.service import Service
@@ -57,6 +58,8 @@ class _AssetsService(ABC):
     icon_types: ClassVar[list[str]]
 
     _client: Optional[AsyncClient] = None
+    _links: dict[str, str] = {}
+    _async_lock: AsyncLock = AsyncLock()
 
     id: int
     type: str
@@ -131,33 +134,72 @@ class _AssetsService(ABC):
                 await file.write(response.content)  # 保存图标
             return path.resolve()
 
-    async def _get_from_ambr(self, item: str) -> Path | None:  # pylint: disable=W0613,R0201
+    async def _get_from_ambr(self, item: str) -> str | None:  # pylint: disable=W0613,R0201
+        """从 ambr.top 上爬取"""
         return None
 
-    async def _get_from_enka(self, item: str) -> Path | None:  # pylint: disable=W0613,R0201
+    async def _get_from_enka(self, item: str) -> str | None:  # pylint: disable=W0613,R0201
+        """从 enke.network 上爬取"""
         return None
 
-    async def _get_from_honey(self, item: str) -> Path | None:
+    async def _get_from_honey(self, item: str) -> str | None:
-        """从 honey 获取图标"""
+        """从 honey 上爬取"""
-        if (url := self.honey_name_map.get(item, None)) is not None:
+        try:
-            # 先尝试下载 png 格式的图片
+            honey_name = self.honey_name_map.get(item, None)
-            path = self.path.joinpath(f"{item}.png")
+        except IndexError:
-            if (result := await self._download(HONEY_HOST.join(f"img/{url}.png"), path)) is not None:
+            return None
+        if honey_name is not None:
+            try:
+                result = HONEY_HOST.join(f"img/{honey_name}.png")
+                response = await self.client.get(result, follow_redirects=False)
+                response.raise_for_status()
+            except HTTPStatusError:
+                return None
+            if response.status_code == 200:
                 return result
-            path = self.path.joinpath(f"{item}.webp")
+
-            return await self._download(HONEY_HOST.join(f"img/{url}.webp"), path)
+            return HONEY_HOST.join(f"img/{honey_name}.webp")
+
+    async def _download_url_generator(self, item: str) -> AsyncIterator[str]:
+        for func in map(lambda x: getattr(self, x), sorted(filter(lambda x: x.startswith("_get_from_"), dir(self)))):
+            if (url := await func(item)) is not None:
+                try:
+                    response = await self.client.get(url := str(url))
+                    response.raise_for_status()
+                    if response.status_code == 200:
+                        yield url
+                except HTTPStatusError:
+                    continue
+
+    async def _get_download_url(self, item: str) -> str | None:
+        """获取图标的下载链接"""
+        async for url in self._download_url_generator(item):
+            if url is not None:
+                return url
 
     async def _get_img(self, overwrite: bool = False, *, item: str) -> Path | None:
         """获取图标"""
         path = next(filter(lambda x: x.stem == item, self.path.iterdir()), None)
-        if not overwrite and path:
+        if not overwrite and path:  # 如果需要下载的图标存在且不覆盖( overwrite )
             return path.resolve()
-        if overwrite and path is not None and path.exists():
+        if path is not None and path.exists():
-            await async_remove(path)
+            if overwrite:  # 如果覆盖
-        # 依次从使用当前 assets class 中的爬虫下载图标，顺序为爬虫名的字母顺序
+                await async_remove(path)  # 删除已存在的图标
-        for func in map(lambda x: getattr(self, x), sorted(filter(lambda x: x.startswith("_get_from_"), dir(self)))):
+            else:
-            if (path := await func(item)) is not None:
                 return path
+        # 依次从使用当前 assets class 中的爬虫下载图标，顺序为爬虫名的字母顺序
+        async for url in self._download_url_generator(item):
+            if url is not None:
+                path = self.path.joinpath(f"{item}{Path(url).suffix}")
+                if (result := await self._download(url, path)) is not None:
+                    return result
+
+    async def get_link(self, item: str) -> str | None:
+        async with self._async_lock:
+            if (result := self._links.get(item, None)) is None:
+                result = await self._get_download_url(item)
+                self._links.update({item: result})
+        return result
 
     def __getattr__(self, item: str):
         """魔法"""
@@ -231,17 +273,15 @@ class _AvatarAssets(_AssetsService):
         result._enka_api = self._enka_api
         return result
 
-    async def _get_from_ambr(self, item: str) -> Path | None:
+    async def _get_from_ambr(self, item: str) -> str | None:
         if item in {"icon", "side", "gacha"}:
-            url = AMBR_HOST.join(f"assets/UI/{self.game_name_map[item]}.png")
+            return str(AMBR_HOST.join(f"assets/UI/{self.game_name_map[item]}.png"))
-            return await self._download(url, self.path.joinpath(f"{item}.png"))
 
-    async def _get_from_enka(self, item: str) -> Path | None:
+    async def _get_from_enka(self, item: str) -> str | None:
-        path = self.path.joinpath(f"{item}.png")
         item = "banner" if item == "gacha" else item
         # noinspection PyUnboundLocalVariable
         if self.enka is not None and item in (data := self.enka.images.dict()).keys() and (url := data[item]["url"]):
-            return await self._download(url, path)
+            return str(url)
 
     @cached_property
     def honey_name_map(self) -> dict[str, str]:
@@ -295,11 +335,13 @@ class _WeaponAssets(_AssetsService):
         result.id = target
         return result
 
-    async def _get_from_enka(self, item: str) -> Path | None:
+    async def _get_from_ambr(self, item: str) -> str | None:
+        if item == "icon":
+            return str(AMBR_HOST.join(f"assets/UI/{self.game_name_map.get(item)}.png"))
+
+    async def _get_from_enka(self, item: str) -> str | None:
         if item in self.game_name_map:
-            url = ENKA_HOST.join(f"ui/{self.game_name_map.get(item)}.png")
+            return str(ENKA_HOST.join(f"ui/{self.game_name_map.get(item)}.png"))
-            path = self.path.joinpath(f"{item}.png")
-            return await self._download(url, path)
 
     @cached_property
     def honey_name_map(self) -> dict[str, str]:
@@ -336,20 +378,21 @@ class _MaterialAssets(_AssetsService):
         result.id = target
         return result
 
-    async def _get_from_ambr(self, item: str) -> Path | None:
+    async def _get_from_ambr(self, item: str) -> str | None:
         if item == "icon":
-            url = AMBR_HOST.join(f"assets/UI/{self.game_name_map.get(item)}.png")
+            return str(AMBR_HOST.join(f"assets/UI/{self.game_name_map.get(item)}.png"))
-            path = self.path.joinpath(f"{item}.png")
-            return await self._download(url, path)
 
-    async def _get_from_honey(self, item: str) -> Path | None:
+    async def _get_from_honey(self, item: str) -> str | None:
-        path = self.path.joinpath(f"{item}.png")
+        try:
-        url = HONEY_HOST.join(f"/img/{self.honey_name_map.get(item)}.png")
+            result = HONEY_HOST.join(f"/img/{self.honey_name_map.get(item)}.png")
-        if (result := await self._download(url, path)) is None:
+            response = await self.client.get(result, follow_redirects=False)
-            path = self.path.joinpath(f"{item}.webp")
+            response.raise_for_status()
-            url = HONEY_HOST.join(f"/img/{self.honey_name_map.get(item)}.webp")
+        except HTTPStatusError:
-            return await self._download(url, path)
+            return None
-        return result
+        if response.status_code == 200:
+            return result
+
+        return HONEY_HOST.join(f"/img/{self.honey_name_map.get(item)}.webp")
 
 
 class _ArtifactAssets(_AssetsService):
@@ -376,16 +419,13 @@ class _ArtifactAssets(_AssetsService):
     def game_name(self) -> str:
         return f"UI_RelicIcon_{self.id}"
 
-    async def _get_from_enka(self, item: str) -> Path | None:
+    async def _get_from_enka(self, item: str) -> str | None:
         if item in self.game_name_map:
-            url = ENKA_HOST.join(f"ui/{self.game_name_map.get(item)}.png")
+            return str(ENKA_HOST.join(f"ui/{self.game_name_map.get(item)}.png"))
-            path = self.path.joinpath(f"{item}.png")
-            return await self._download(url, path)
 
-    async def _get_from_ambr(self, item: str) -> Path | None:
+    async def _get_from_ambr(self, item: str) -> str | None:
         if item in self.game_name_map:
-            url = AMBR_HOST.join(f"assets/UI/reliquary/{self.game_name_map[item]}.png")
+            return str(AMBR_HOST.join(f"assets/UI/reliquary/{self.game_name_map[item]}.png"))
-            return await self._download(url, self.path.joinpath(f"{item}.png"))
 
     @cached_property
     def game_name_map(self) -> dict[str, str]:
@@ -434,16 +474,14 @@ class _NamecardAssets(_AssetsService):
         result.enka = DEFAULT_EnkaAssets.namecards(target)
         return result
 
-    async def _get_from_ambr(self, item: str) -> Path | None:
+    async def _get_from_ambr(self, item: str) -> str | None:
         if item == "profile":
-            url = AMBR_HOST.join(f"assets/UI/namecard/{self.game_name_map[item]}.png.png")
+            return AMBR_HOST.join(f"assets/UI/namecard/{self.game_name_map[item]}.png.png")
-            return await self._download(url, self.path.joinpath(f"{item}.png"))
 
-    async def _get_from_enka(self, item: str) -> Path | None:
+    async def _get_from_enka(self, item: str) -> str | None:
-        path = self.path.joinpath(f"{item}.png")
         url = getattr(self.enka, {"profile": "banner"}.get(item, item), None)
         if url is not None:
-            return await self._download(url.url, path)
+            return str(url)
 
     @cached_property
     def game_name_map(self) -> dict[str, str]:

metadata/genshin.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import functools
-from typing import Any, Generic, ItemsView, Iterator, KeysView, TypeVar, Optional, ValuesView
+from typing import Any, Generic, ItemsView, Iterator, KeysView, Optional, TypeVar, ValuesView
 
 import ujson as json
 
@@ -60,6 +60,12 @@ class Data(dict, Generic[K, V]):
         self._dict = {}
         super(Data, self).__init__()
 
+    def __str__(self) -> str:
+        return self.data.__str__()
+
+    def __repr__(self) -> str:
+        return self.data.__repr__()
+
     def get(self, key: K, value: Any = None) -> V | None:
         return self.data.get(key, value)