from __future__ import annotations import asyncio import re from typing import Dict, List, Optional import ujson as json from aiofiles import open as async_open from httpx import AsyncClient, HTTPError, Response from modules.wiki.base import HONEY_HOST from utils.const import PROJECT_ROOT from utils.log import logger from utils.typedefs import StrOrInt __all__ = [ "get_avatar_data", "get_artifact_data", "get_material_data", "get_namecard_data", "get_weapon_data", "update_honey_metadata", ] DATA_TYPE = Dict[StrOrInt, List[str]] FULL_DATA_TYPE = Dict[str, DATA_TYPE] client = AsyncClient() async def request(url: str, retry: int = 5) -> Optional[Response]: for time in range(retry): try: return await client.get(url) except HTTPError: if time != retry - 1: await asyncio.sleep(1) continue return None except Exception as e: raise e async def get_avatar_data() -> DATA_TYPE: result = {} url = "https://genshin.honeyhunterworld.com/fam_chars/?lang=CHS" response = await request(url) chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0] json_data = json.loads(chaos_data) # 转为 json for data in json_data: cid = int("10000" + re.findall(r"\d+", data[1])[0]) honey_id = re.findall(r"/(.*?)/", data[1])[0] name = re.findall(r">(.*)<", data[1])[0] rarity = int(re.findall(r">(\d)<", data[2])[0]) result[cid] = [honey_id, name, rarity] return result async def get_weapon_data() -> DATA_TYPE: from modules.wiki.other import WeaponType result = {} urls = [HONEY_HOST.join(f"fam_{i.lower()}/?lang=CHS") for i in WeaponType.__members__] for url in urls: response = await request(url) chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0] json_data = json.loads(chaos_data) # 转为 json for data in json_data: name = re.findall(r">(.*)<", data[1])[0] if name in ["「一心传」名刀", "石英大剑", "琥珀玥", "黑檀弓"]: # 跳过特殊的武器 continue wid = int(re.findall(r"\d+", data[1])[0]) honey_id = re.findall(r"/(.*?)/", data[1])[0] rarity = int(re.findall(r">(\d)<", data[2])[0]) result[wid] = [honey_id, name, rarity] return result async def get_material_data() -> DATA_TYPE: result = {} weapon = [HONEY_HOST.join(f"fam_wep_{i}/?lang=CHS") for i in ["primary", "secondary", "common"]] talent = [HONEY_HOST.join(f"fam_talent_{i}/?lang=CHS") for i in ["book", "boss", "common", "reward"]] namecard = [HONEY_HOST.join("fam_nameplate/?lang=CHS")] urls = weapon + talent + namecard response = await request("https://api.ambr.top/v2/chs/material") ambr_data = json.loads(response.text)["data"]["items"] for url in urls: response = await request(url) chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0] json_data = json.loads(chaos_data) # 转为 json for data in json_data: honey_id = re.findall(r"/(.*?)/", data[1])[0] name = re.findall(r">(.*)<", data[1])[0] rarity = int(re.findall(r">(\d)<", data[2])[0]) mid = None for mid, item in ambr_data.items(): if name == item["name"]: break mid = int(mid) or int(re.findall(r"\d+", data[1])[0]) result[mid] = [honey_id, name, rarity] return result async def get_artifact_data() -> DATA_TYPE: async def get_first_id(_link) -> str: _response = await request(_link) _chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", _response.text)[0] _json_data = json.loads(_chaos_data) return re.findall(r"/(.*?)/", _json_data[-1][1])[0] result = {} url = "https://genshin.honeyhunterworld.com/fam_art_set/?lang=CHS" response = await request("https://api.ambr.top/v2/chs/reliquary") ambr_data = json.loads(response.text)["data"]["items"] response = await request(url) chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0] json_data = json.loads(chaos_data) # 转为 json for data in json_data: honey_id = re.findall(r"/(.*?)/", data[1])[0] name = re.findall(r"alt=\"(.*?)\"", data[0])[0] link = HONEY_HOST.join(re.findall(r'href="(.*?)"', data[0])[0]) first_id = await get_first_id(link) aid = None for aid, item in ambr_data.items(): if name == item["name"]: break aid = aid or re.findall(r"\d+", data[1])[0] result[aid] = [honey_id, name, first_id] return result async def get_namecard_data() -> DATA_TYPE: from metadata.genshin import NAMECARD_DATA if not NAMECARD_DATA: # noinspection PyProtectedMember from metadata.genshin import Data from metadata.scripts.metadatas import update_metadata_from_github await update_metadata_from_github() # noinspection PyPep8Naming NAMECARD_DATA = Data("namecard") url = HONEY_HOST.join("fam_nameplate/?lang=CHS") result = {} response = await request(url) chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0] json_data = json.loads(chaos_data) for data in json_data: honey_id = re.findall(r"/(.*?)/", data[1])[0] name = re.findall(r"alt=\"(.*?)\"", data[0])[0] try: nid = [key for key, value in NAMECARD_DATA.items() if value["name"] == name][0] except IndexError: # 暂不支持 beta 的名片 continue rarity = int(re.findall(r">(\d)<", data[2])[0]) result[nid] = [honey_id, name, rarity] return result async def update_honey_metadata(overwrite: bool = True) -> FULL_DATA_TYPE | None: path = PROJECT_ROOT.joinpath("metadata/data/honey.json") if not overwrite and path.exists(): return avatar_data = await get_avatar_data() logger.success("Avatar data is done.") weapon_data = await get_weapon_data() logger.success("Weapon data is done.") material_data = await get_material_data() logger.success("Material data is done.") artifact_data = await get_artifact_data() logger.success("Artifact data is done.") namecard_data = await get_namecard_data() logger.success("Namecard data is done.") result = { "avatar": avatar_data, "weapon": weapon_data, "material": material_data, "artifact": artifact_data, "namecard": namecard_data, } path.parent.mkdir(parents=True, exist_ok=True) async with async_open(path, mode="w", encoding="utf-8") as file: await file.write(json.dumps(result, ensure_ascii=False, indent=4)) return result