开箱新增更新指定刀具皮肤命令

plugins/open_cases/utils.py

@@ -1,9 +1,10 @@
 import asyncio
 import os
 import random
+import re
 import time
 from datetime import datetime
-from typing import List, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 import nonebot
 from tortoise.functions import Count
@@ -16,13 +17,21 @@ from utils.image_utils import BuildImage
 from utils.utils import broadcast_group, cn2py
 
 from .build_image import generate_skin
-from .config import CASE2ID, CASE_BACKGROUND, COLOR2NAME, NAME2COLOR
+from .config import (
+    CASE2ID,
+    CASE_BACKGROUND,
+    COLOR2NAME,
+    KNIFE2ID,
+    NAME2COLOR,
+    UpdateType,
+)
 from .models.buff_skin import BuffSkin
 from .models.buff_skin_log import BuffSkinLog
 from .models.open_cases_user import OpenCasesUser
 
 URL = "https://buff.163.com/api/market/goods"
-# proxies = 'http://49.75.59.242:3128'
+
+SELL_URL = "https://buff.163.com/goods"
 
 
 driver = nonebot.get_driver()
@@ -41,44 +50,80 @@ class CaseManager:
         )
 
 
-async def update_case_data(case_name: str) -> str:
-    """更新皮肤数据
+async def update_skin_data(name: str) -> str:
+    """更新箱子内皮肤数据
 
     Args:
-        case_name (str): 箱子名称
+        name (str): 箱子名称
 
     Returns:
         _type_: _description_
     """
-    if case_name not in CASE2ID:
-        return "未在当前指定武器箱捏"
+    type_ = None
+    if name in CASE2ID:
+        type_ = UpdateType.CASE
+    if name in KNIFE2ID:
+        type_ = UpdateType.WEAPON_TYPE
+    if not type_:
+        return "未在指定武器箱或指定武器类型内"
     session = Config.get_config("open_cases", "COOKIE")
     if not session:
         return "BUFF COOKIE为空捏!"
-    db_skin_list = await BuffSkin.filter(case_name=case_name).all()
-    db_skin_name_list = [
-        skin.name + skin.skin_name + skin.abrasion for skin in db_skin_list
-    ]
-    data_list, total = await search_skin_page(case_name, 1)
+    weapon2case = {}
+    if type_ == UpdateType.CASE:
+        db_skin_id_list = [
+            skin.skin_id for skin in await BuffSkin.filter(case_name=name).all()
+        ]
+    else:
+        db_data = await BuffSkin.filter(name__contains=name).all()
+        db_skin_id_list = [
+            skin.skin_id for skin in await BuffSkin.filter(name__contains=name).all()
+        ]
+        weapon2case = {
+            item.name + item.skin_name: item.case_name
+            for item in db_data
+            if item.case_name != "未知武器箱"
+        }
+    data_list, total = await search_skin_page(name, 1, type_)
     if isinstance(data_list, str):
         return data_list
     for page in range(2, total + 1):
-        rand_time = random.randint(10, 50)
+        rand_time = random.randint(20, 50)
         logger.debug(f"访问随机等待时间: {rand_time}", "开箱更新")
         await asyncio.sleep(rand_time)
-        data_list_, total = await search_skin_page(case_name, page)
+        data_list_, total = await search_skin_page(name, page, type_)
         if isinstance(data_list_, list):
             data_list += data_list_
     create_list: List[BuffSkin] = []
     update_list: List[BuffSkin] = []
     log_list = []
-    case_name_py = cn2py(case_name)
     now = datetime.now()
+    exists_id_list = []
     for skin in data_list:
-        name = skin.name + skin.skin_name + skin.abrasion
+        if skin.skin_id in exists_id_list:
+            continue
+        exists_id_list.append(skin.skin_id)
+        key = skin.name + skin.skin_name
+        name_ = skin.name + skin.skin_name + skin.abrasion
         skin.create_time = now
         skin.update_time = now
-        if name in db_skin_name_list:
+        if not skin.case_name:
+            case_name = weapon2case.get(key)
+            if not case_name:
+                case_name = await get_skin_case(skin.skin_id)
+                rand = random.randint(10, 20)
+                logger.debug(
+                    f"获取 {skin.name} | {skin.skin_name} 皮肤所属武器箱: {case_name}, 访问随机等待时间: {rand}",
+                    "开箱更新",
+                )
+                await asyncio.sleep(rand)
+            if not case_name:
+                case_name = "未知武器箱"
+            else:
+                weapon2case[key] = case_name
+            case_name = case_name.replace("”", "").replace("“", "")
+            skin.case_name = case_name
+        if skin.skin_id in db_skin_id_list:
             update_list.append(skin)
         else:
             create_list.append(skin)
@@ -100,8 +145,8 @@ async def update_case_data(case_name: str) -> str:
                 create_time=now,
             )
         )
-        name = skin.name + "-" + skin.skin_name + "-" + skin.abrasion
-        file_path = BASE_PATH / case_name_py / f"{cn2py(name)}.jpg"
+        name_ = skin.name + "-" + skin.skin_name + "-" + skin.abrasion
+        file_path = BASE_PATH / cn2py(skin.case_name) / f"{cn2py(name_)}.jpg"
         if not file_path.exists():
             logger.debug(f"下载皮肤 {name} 图片: {skin.img_url}...", "开箱更新")
             await AsyncHttpx.download_file(skin.img_url, file_path)
@@ -109,9 +154,9 @@ async def update_case_data(case_name: str) -> str:
             await asyncio.sleep(rand_time)
             logger.debug(f"图片下载随机等待时间: {rand_time}", "开箱更新")
         else:
-            logger.debug(f"皮肤 {name} 图片已存在...", "开箱更新")
+            logger.debug(f"皮肤 {name_} 图片已存在...", "开箱更新")
     if create_list:
-        logger.debug(f"更新武器箱: [<u><e>{case_name}</e></u>], 创建 {len(create_list)} 个皮肤!")
+        logger.debug(f"更新武器箱/皮肤: [<u><e>{name}</e></u>], 创建 {len(create_list)} 个皮肤!")
         await BuffSkin.bulk_create(create_list, 10)
     if update_list:
         abrasion_list = []
@@ -125,7 +170,7 @@ async def update_case_data(case_name: str) -> str:
             if skin.skin_name not in skin_name_list:
                 skin_name_list.append(skin.skin_name)
         db_data = await BuffSkin.filter(
-            case_name=case_name,
+            case_name=name,
             skin_name__in=skin_name_list,
             name__in=name_list,
             abrasion__in=abrasion_list,
@@ -146,7 +191,7 @@ async def update_case_data(case_name: str) -> str:
                     data.sell_reference_price = skin.sell_reference_price
                     data.update_time = skin.update_time
                     _update_list.append(data)
-        logger.debug(f"更新武器箱: [<u><c>{case_name}</c></u>], 更新 {len(create_list)} 个皮肤!")
+        logger.debug(f"更新武器箱/皮肤: [<u><c>{name}</c></u>], 更新 {len(create_list)} 个皮肤!")
         await BuffSkin.bulk_update(
             _update_list,
             [
@@ -161,105 +206,129 @@ async def update_case_data(case_name: str) -> str:
             10,
         )
     if log_list:
-        logger.debug(f"更新武器箱: [<u><e>{case_name}</e></u>], 新增 {len(log_list)} 条皮肤日志!")
+        logger.debug(f"更新武器箱/皮肤: [<u><e>{name}</e></u>], 新增 {len(log_list)} 条皮肤日志!")
         await BuffSkinLog.bulk_create(log_list)
-    if case_name not in CaseManager.CURRENT_CASES:
+    if name not in CaseManager.CURRENT_CASES:
         CaseManager.CURRENT_CASES.append(case_name)  # type: ignore
-    return f"更新武器箱: [{case_name}] 成功, 共更新 {len(update_list)} 个皮肤, 新创建 {len(create_list)} 个皮肤!"
+    return f"更新武器箱/皮肤: [{name}] 成功, 共更新 {len(update_list)} 个皮肤, 新创建 {len(create_list)} 个皮肤!"
 
 
 async def search_skin_page(
-    case_name: str, page_index: int
+    name: str, page_index: int, type_: UpdateType
 ) -> Tuple[Union[List[BuffSkin], str], int]:
     """查询箱子皮肤
 
     Args:
-        case_name (str): 箱子名称
+        name (str): 箱子名称
         page_index (int): 页数
 
     Returns:
         Union[List[BuffSkin], str]: BuffSkin
     """
     logger.debug(
-        f"尝试访问武器箱: [<u><e>{case_name}</e></u>] 页数: [<u><y>{page_index}</y></u>]", "开箱更新"
+        f"尝试访问武器箱/皮肤: [<u><e>{name}</e></u>] 页数: [<u><y>{page_index}</y></u>]", "开箱更新"
     )
     cookie = {"session": Config.get_config("open_cases", "COOKIE")}
     params = {
         "game": "csgo",
         "page_num": page_index,
         "page_size": 80,
-        "itemset": CASE2ID[case_name],
         "_": time.time(),
         "use_suggestio": 0,
     }
+    if type_ == UpdateType.CASE:
+        params["itemset"] = CASE2ID[name]
+    elif type_ == UpdateType.WEAPON_TYPE:
+        params["category"] = KNIFE2ID[name]
     proxy = None
     if ip := Config.get_config("open_cases", "BUFF_PROXY"):
         proxy = {"http://": ip, "https://": ip}
-    response = await AsyncHttpx.get(
-        URL,
-        proxy=proxy,
-        params=params,
-        cookies=cookie,  # type: ignore
-    )
-    logger.debug(f"访问BUFF API: {response.text}", "更新武器箱")
-    json_data = response.json()
-    update_data = []
-    if json_data["code"] == "OK":
-        data_list = json_data["data"]["items"]
-        for data in data_list:
-            obj = {"case_name": case_name}
-            name = data["name"]
-            try:
-                logger.debug(
-                    f"武器箱: [<u><e>{case_name}</e></u>] 页数: [<u><y>{page_index}</y></u>] 正在收录皮肤: [<u><c>{name}</c></u>]...",
-                    "开箱更新",
-                )
-                obj["buy_max_price"] = data["buy_max_price"]  # 求购最大金额
-                obj["buy_num"] = data["buy_num"]  # 当前求购
-                goods_info = data["goods_info"]
-                info = goods_info["info"]
-                tags = info["tags"]
-                obj["weapon_type"] = tags["type"]["localized_name"]  # 枪械类型
-                if obj["weapon_type"] in ["音乐盒", "印花", "探员"]:
-                    continue
-                elif obj["weapon_type"] in ["匕首", "手套"]:
-                    obj["color"] = "KNIFE"
-                    obj["name"] = data["short_name"].split("（")[0].strip()  # 名称
-                elif obj["weapon_type"] in ["武器箱"]:
-                    obj["color"] = "CASE"
-                    obj["name"] = data["short_name"]
-                else:
-                    obj["color"] = NAME2COLOR[tags["rarity"]["localized_name"]]
-                    obj["name"] = tags["weapon"]["localized_name"]  # 名称
-                if obj["weapon_type"] not in ["武器箱"]:
-                    obj["abrasion"] = tags["exterior"]["localized_name"]  # 磨损
-                    obj["is_stattrak"] = "StatTrak" in tags["quality"]["localized_name"]  # type: ignore # 是否暗金
-                    if not obj["color"]:
-                        obj["color"] = NAME2COLOR[
-                            tags["rarity"]["localized_name"]
-                        ]  # 品质颜色
-                else:
-                    obj["abrasion"] = "CASE"
-                obj["skin_name"] = data["short_name"].split("|")[-1].strip()  # 皮肤名称
-                obj["img_url"] = goods_info["original_icon_url"]  # 图片url
-                obj["steam_price"] = goods_info["steam_price_cny"]  # steam价格
-                obj["sell_min_price"] = data["sell_min_price"]  # 售卖最低价格
-                obj["sell_num"] = data["sell_num"]  # 售卖数量
-                obj["sell_reference_price"] = data["sell_reference_price"]  # 参考价格
-                update_data.append(BuffSkin(**obj))
-            except Exception as e:
-                logger.error(
-                    f"更新武器箱: [<u><e>{case_name}</e></u>] 皮肤: [<u><c>{name}</c></u>] 错误",
-                    e=e,
-                )
-        logger.debug(
-            f"访问武器箱: [<u><e>{case_name}</e></u>] 页数: [<u><y>{page_index}</y></u>] 成功并收录完成",
-            "开箱更新",
-        )
-        return update_data, json_data["data"]["total_page"]
-    else:
-        logger.warning(f'访问BUFF失败: {json_data["error"]}')
-    return f'访问失败: {json_data["error"]}', -1
+    response = None
+    error = ""
+    for i in range(3):
+        try:
+            response = await AsyncHttpx.get(
+                URL,
+                proxy=proxy,
+                params=params,
+                cookies=cookie,  # type: ignore
+            )
+            if response.status_code == 200:
+                break
+            rand = random.randint(3, 7)
+            logger.debug(
+                f"尝试访问武器箱/皮肤第 {i+1} 次访问异常, code: {response.status_code}", "开箱更新"
+            )
+            await asyncio.sleep(rand)
+        except Exception as e:
+            logger.debug(f"尝试访问武器箱/皮肤第 {i+1} 次访问发生错误 {type(e)}: {e}", "开箱更新")
+            error = f"{type(e)}: {e}"
+    if not response:
+        return f"访问发生异常: {error}", -1
+    if response.status_code == 200:
+        logger.debug(f"访问BUFF API: {response.text}", "开箱更新")
+        json_data = response.json()
+        update_data = []
+        if json_data["code"] == "OK":
+            data_list = json_data["data"]["items"]
+            for data in data_list:
+                obj = {}
+                if type_ == UpdateType.CASE:
+                    obj["case_name"] = name
+                name = data["name"]
+                try:
+                    logger.debug(
+                        f"武器箱: [<u><e>{name}</e></u>] 页数: [<u><y>{page_index}</y></u>] 正在收录皮肤: [<u><c>{name}</c></u>]...",
+                        "开箱更新",
+                    )
+                    obj["skin_id"] = str(data["id"])
+                    obj["buy_max_price"] = data["buy_max_price"]  # 求购最大金额
+                    obj["buy_num"] = data["buy_num"]  # 当前求购
+                    goods_info = data["goods_info"]
+                    info = goods_info["info"]
+                    tags = info["tags"]
+                    obj["weapon_type"] = tags["type"]["localized_name"]  # 枪械类型
+                    if obj["weapon_type"] in ["音乐盒", "印花", "探员"]:
+                        continue
+                    elif obj["weapon_type"] in ["匕首", "手套"]:
+                        obj["color"] = "KNIFE"
+                        obj["name"] = data["short_name"].split("（")[0].strip()  # 名称
+                    elif obj["weapon_type"] in ["武器箱"]:
+                        obj["color"] = "CASE"
+                        obj["name"] = data["short_name"]
+                    else:
+                        obj["color"] = NAME2COLOR[tags["rarity"]["localized_name"]]
+                        obj["name"] = tags["weapon"]["localized_name"]  # 名称
+                    if obj["weapon_type"] not in ["武器箱"]:
+                        obj["abrasion"] = tags["exterior"]["localized_name"]  # 磨损
+                        obj["is_stattrak"] = "StatTrak" in tags["quality"]["localized_name"]  # type: ignore # 是否暗金
+                        if not obj["color"]:
+                            obj["color"] = NAME2COLOR[
+                                tags["rarity"]["localized_name"]
+                            ]  # 品质颜色
+                    else:
+                        obj["abrasion"] = "CASE"
+                    obj["skin_name"] = data["short_name"].split("|")[-1].strip()  # 皮肤名称
+                    obj["img_url"] = goods_info["original_icon_url"]  # 图片url
+                    obj["steam_price"] = goods_info["steam_price_cny"]  # steam价格
+                    obj["sell_min_price"] = data["sell_min_price"]  # 售卖最低价格
+                    obj["sell_num"] = data["sell_num"]  # 售卖数量
+                    obj["sell_reference_price"] = data["sell_reference_price"]  # 参考价格
+                    update_data.append(BuffSkin(**obj))
+                except Exception as e:
+                    logger.error(
+                        f"更新武器箱: [<u><e>{name}</e></u>] 皮肤: [<u><c>{name}</c></u>] 错误",
+                        e=e,
+                    )
+            logger.debug(
+                f"访问武器箱: [<u><e>{name}</e></u>] 页数: [<u><y>{page_index}</y></u>] 成功并收录完成",
+                "开箱更新",
+            )
+            return update_data, json_data["data"]["total_page"]
+        else:
+            logger.warning(f'访问BUFF失败: {json_data["error"]}')
+        return f'访问失败: {json_data["error"]}', -1
+    return f"访问失败, 状态码: {response.status_code}", -1
 
 
 async def build_case_image(case_name: str) -> Union[BuildImage, str]:
@@ -421,6 +490,32 @@ def get_bk_image_size(
     return new_size
 
 
+async def get_skin_case(id_: str) -> Optional[str]:
+    """获取皮肤所在箱子
+
+    Args:
+        id_ (str): 皮肤id
+
+    Returns:
+        Optional[str]: 武器箱名称
+    """
+    url = f"{SELL_URL}/{id_}"
+    proxy = None
+    if ip := Config.get_config("open_cases", "BUFF_PROXY"):
+        proxy = {"http://": ip, "https://": ip}
+    response = await AsyncHttpx.get(
+        url,
+        proxy=proxy,
+    )
+    if response.status_code == 200:
+        text = response.text
+        if r := re.search('<meta name="description".*,(.*)武器箱.*?>', text):
+            return r.group(1)
+    else:
+        logger.debug(f"访问皮肤所属武器箱异常 url: {url} code: {response.status_code}")
+    return None
+
+
 async def reset_count_daily():
     """
     重置每日开箱
@@ -437,57 +532,3 @@ async def reset_count_daily():
 @driver.on_startup
 async def _():
     await CaseManager.reload()
-
-
-@driver.on_startup
-async def _():
-    """
-    将旧表数据移动到新表
-    """
-    # if not await BuffSkin.first() and await BuffPrice.first():
-    #     logger.debug("开始移动旧表数据 BuffPrice -> BuffSkin")
-    #     id2name = {1: "狂牙大行动", 2: "突围大行动", 3: "命悬一线", 4: "裂空", 5: "光谱"}
-    #     data_list: List[BuffSkin] = []
-    #     for data in await BuffPrice.all():
-    #         logger.debug(f"移动旧表数据: {data.skin_name}")
-    #         case_name = id2name[data.case_id]
-    #         name = data.skin_name
-    #         is_stattrak = "StatTrak" in name
-    #         name = name.replace("（★ StatTrak™）", "").replace("（StatTrak™）", "").strip()
-    #         name, skin_name = name.split("|")
-    #         abrasion = "无涂装"
-    #         if "(" in skin_name:
-    #             skin_name, abrasion = skin_name.split("(")
-    #             if abrasion.endswith(")"):
-    #                 abrasion = abrasion[:-1]
-    #         color = get_color(case_name, name.strip(), skin_name.strip())
-    #         if not color:
-    #             search_list = [
-    #                 x
-    #                 for x in data_list
-    #                 if x.skin_name == skin_name.strip() and x.name == name.strip()
-    #             ]
-    #             if search_list:
-    #                 color = get_color(
-    #                     case_name, search_list[0].name, search_list[0].skin_name
-    #                 )
-    #             if not color:
-    #                 logger.debug(
-    #                     f"箱子: [{case_name}] 皮肤: [{name}|{skin_name}] 未获取到皮肤品质，跳过..."
-    #                 )
-    #                 continue
-    #         data_list.append(
-    #             BuffSkin(
-    #                 case_name=case_name,
-    #                 name=name.strip(),
-    #                 skin_name=skin_name.strip(),
-    #                 is_stattrak=is_stattrak,
-    #                 abrasion=abrasion.strip(),
-    #                 skin_price=data.skin_price,
-    #                 color=color,
-    #                 create_time=datetime.now(),
-    #                 update_time=datetime.now(),
-    #             )
-    #         )
-    #     await BuffSkin.bulk_create(data_list, batch_size=10)
-    #     logger.debug("完成移动旧表数据 BuffPrice -> BuffSkin")