✨实现资源自动下载

StarRailUID/starrailuid_charinfo/draw_char_img.py

@@ -22,7 +22,7 @@ from ..utils.resource.RESOURCE_PATH import (
     SKILL_PATH,
     PLAYER_PATH,
     WEAPON_PATH,
-    CHAR_PORTRAIT,
+    CHAR_PORTRAIT_PATH,
 )
 from ..utils.fonts.starrail_fonts import (
     sr_font_20,
@@ -83,7 +83,7 @@ async def draw_char_info_img(raw_mes: str, sr_uid: str, url: Optional[str]):
 
     # 放角色立绘
     char_info = bg_img.copy()
-    char_img = Image.open(CHAR_PORTRAIT / f'{char.char_id}.png').resize(
+    char_img = Image.open(CHAR_PORTRAIT_PATH / f'{char.char_id}.png').resize(
         (1050, 1050)
     )
     char_info.paste(char_img, (-220, -130), char_img)

StarRailUID/starrailuid_resource/__init__.py

@@ -0,0 +1,20 @@
+from gsuid_core.sv import SV
+from gsuid_core.bot import Bot
+from gsuid_core.models import Event
+from gsuid_core.logger import logger
+
+from ..utils.resource.download_all_resource import download_all_resource
+
+sv_sr_download_config = SV('sr下载资源')
+
+
+@sv_sr_download_config.on_fullmatch(('sr下载全部资源'))
+async def send_download_resource_msg(bot: Bot, ev: Event):
+    await bot.send('sr正在开始下载~可能需要较久的时间!')
+    im = await download_all_resource()
+    await bot.send(im)
+
+
+async def startup():
+    logger.info('[sr资源文件下载] 正在检查与下载缺失的资源文件，可能需要较长时间，请稍等')
+    logger.info(f'[sr资源文件下载] {await download_all_resource()}')

StarRailUID/utils/resource/RESOURCE_PATH.py

@@ -5,32 +5,39 @@ from gsuid_core.data_store import get_res_path
 
 MAIN_PATH = get_res_path() / 'StarRailUID'
 sys.path.append(str(MAIN_PATH))
-CONFIG_PATH = MAIN_PATH / 'config.json'
-RESOURCE_PATH = MAIN_PATH / 'resource'
-PLAYER_PATH = MAIN_PATH / 'players'
+
 CU_BG_PATH = MAIN_PATH / 'bg'
-TEMP_PATH = RESOURCE_PATH / 'temp'
+CONFIG_PATH = MAIN_PATH / 'config.json'
+PLAYER_PATH = MAIN_PATH / 'players'
+RESOURCE_PATH = MAIN_PATH / 'resource'
+
 CHAR_ICON_PATH = RESOURCE_PATH / 'character'
+CHAR_PORTRAIT_PATH = RESOURCE_PATH / 'character_portrait'
+CONSUMABLE_PATH = RESOURCE_PATH / 'consumable'
+ELEMENT_PATH = RESOURCE_PATH / 'element'
 WEAPON_PATH = RESOURCE_PATH / 'light_cone'
-CHAR_PORTRAIT = RESOURCE_PATH / 'character_portrait'
-SKILL_PATH = RESOURCE_PATH / 'skill'
 RELIC_PATH = RESOURCE_PATH / 'relic'
+SKILL_PATH = RESOURCE_PATH / 'skill'
+TEMP_PATH = RESOURCE_PATH / 'temp'
+
 TEXT2D_PATH = Path(__file__).parent / 'texture2d'
 
 
 def init_dir():
     for i in [
         MAIN_PATH,
-        RESOURCE_PATH,
-        PLAYER_PATH,
-        CHAR_ICON_PATH,
-        WEAPON_PATH,
-        TEXT2D_PATH,
         CU_BG_PATH,
-        TEMP_PATH,
-        CHAR_PORTRAIT,
-        SKILL_PATH,
+        PLAYER_PATH,
+        RESOURCE_PATH,
+        CHAR_ICON_PATH,
+        CHAR_PORTRAIT_PATH,
+        CONSUMABLE_PATH,
+        ELEMENT_PATH,
+        WEAPON_PATH,
         RELIC_PATH,
+        SKILL_PATH,
+        TEXT2D_PATH,
+        TEMP_PATH,
     ]:
         i.mkdir(parents=True, exist_ok=True)
 

StarRailUID/utils/resource/download_all_resource.py

@@ -0,0 +1,11 @@
+import asyncio
+
+from .download_from_cos import download_all_file_from_cos
+
+
+async def download_all_resource():
+    ret = await asyncio.gather(download_all_file_from_cos())
+    ret = [str(x) for x in ret if x]
+    if ret:
+        return '\n'.join(ret)
+    return 'sr全部资源下载完成!'

StarRailUID/utils/resource/download_from_cos.py

@@ -0,0 +1,91 @@
+import os
+import asyncio
+from pathlib import Path
+from typing import Dict, List, Tuple, Union
+
+from msgspec import json as msgjson
+from gsuid_core.logger import logger
+from aiohttp.client import ClientSession
+
+from .download_url import download_file
+from .RESOURCE_PATH import RESOURCE_PATH
+
+with open(
+    Path(__file__).parent / 'resource_map.json', 'r', encoding='UTF-8'
+) as f:
+    resource_map = msgjson.decode(
+        f.read(), type=Dict[str, Dict[str, Dict[str, Union[int, str]]]]
+    )
+
+
+async def download_all_file_from_cos():
+    async def _download(tasks: List[asyncio.Task]):
+        failed_list.extend(
+            list(filter(lambda x: x is not None, await asyncio.gather(*tasks)))
+        )
+        tasks.clear()
+        logger.info('[cos]下载完成!')
+
+    failed_list: List[Tuple[str, int, str]] = []
+    TASKS = []
+    async with ClientSession() as sess:
+        for resource_type in [
+            'character',
+            'character_portrait',
+            'consumable',
+            'element',
+            'light_cone',
+            'relic',
+            'skill',
+        ]:
+            file_dict = resource_map[resource_type]
+            logger.info(f'[cos]数据库[{resource_type}]中存在{len(file_dict)}个内容!')
+            temp_num = 0
+            for file_name, file_info in file_dict.items():
+                name = file_name
+                size = file_info['size']
+                url = file_info['url']
+                path = Path(RESOURCE_PATH / resource_type / name)
+                if path.exists():
+                    is_diff = size == str(os.stat(path).st_size)
+                else:
+                    is_diff = True
+                if (
+                    not path.exists()
+                    or not os.stat(path).st_size
+                    or not is_diff
+                ):
+                    logger.info(f'[cos]开始下载[{resource_type}]_[{name}]...')
+                    temp_num += 1
+                    TASKS.append(
+                        asyncio.wait_for(
+                            download_file(url, resource_type, name, sess),
+                            timeout=60,
+                        )
+                    )
+                    # await download_file(url, FILE_TO_PATH[file], name)
+                    if len(TASKS) >= 10:
+                        await _download(TASKS)
+            else:
+                await _download(TASKS)
+            if temp_num == 0:
+                im = f'[cos]数据库[{resource_type}]无需下载!'
+            else:
+                im = f'[cos]数据库[{resource_type}]已下载{temp_num}个内容!'
+            temp_num = 0
+            logger.info(im)
+    if failed_list:
+        logger.info(f'[cos]开始重新下载失败的{len(failed_list)}个文件...')
+        for url, file, name in failed_list:
+            TASKS.append(
+                asyncio.wait_for(
+                    download_file(url, file, name, sess),
+                    timeout=60,
+                )
+            )
+            if len(TASKS) >= 10:
+                await _download(TASKS)
+        else:
+            await _download(TASKS)
+        if count := len(failed_list):
+            logger.error(f'[cos]仍有{count}个文件未下载，请使用命令 `下载全部资源` 重新下载')

StarRailUID/utils/resource/download_url.py

@@ -0,0 +1,50 @@
+from typing import Tuple, Optional
+
+import aiofiles
+from gsuid_core.logger import logger
+from aiohttp.client import ClientSession
+from aiohttp.client_exceptions import ClientConnectorError
+
+from .RESOURCE_PATH import RESOURCE_PATH
+
+
+async def download(
+    url: str,
+    resource_type: str,
+    name: str,
+) -> Optional[Tuple[str, int, str]]:
+    """
+    :说明:
+      下载URL保存入目录
+    :参数:
+      * url: `str`
+            资源下载地址。
+      * resource_type: `str`
+            资源文件夹名
+      * name: `str`
+            资源保存名称
+    :返回(失败才会有返回值):
+        url: `str`
+        resource_type: `str`
+        name: `str`
+    """
+    async with ClientSession() as sess:
+        return await download_file(url, resource_type, name, sess)
+
+
+async def download_file(
+    url: str,
+    resource_type: str,
+    name: str,
+    sess: Optional[ClientSession] = None,
+) -> Optional[Tuple[str, str, str]]:
+    if sess is None:
+        sess = ClientSession()
+    try:
+        async with sess.get(url) as res:
+            content = await res.read()
+    except ClientConnectorError:
+        logger.warning(f"[cos]{name}下载失败")
+        return url, resource_type, name
+    async with aiofiles.open(RESOURCE_PATH / resource_type / name, "wb") as f:
+        await f.write(content)