From 1e01b141f5bff4dc4e8af7c87ecab5518384eb7f Mon Sep 17 00:00:00 2001 From: xtaodada Date: Tue, 29 Aug 2023 13:15:55 +0800 Subject: [PATCH] refactor: remove almost bbs wiki --- fix_data.py | 4 +- func/README.md | 3 - func/client.py | 41 +++-- func/data.py | 104 +++++++++++++ func/fetch_all.py | 6 +- func/fetch_avatars.py | 45 +----- func/fetch_light_cones.py | 141 ++++-------------- func/fetch_materials.py | 131 ++++++---------- func/fetch_relics.py | 78 +++------- func/url.py | 1 - main.py | 79 ++-------- models/avatar.py | 4 +- models/light_cone.py | 48 ++---- models/material.py | 32 +++- models/monster.py | 25 ---- models/relic.py | 9 +- modules/apihelper/client/base/httpxrequest.py | 10 +- .../apihelper/client/base/hyperionrequest.py | 38 ++++- .../apihelper/client/components/hyperion.py | 58 +++++-- modules/apihelper/error.py | 6 +- res_func/avatar.py | 27 ++-- res_func/light_cone.py | 21 ++- res_func/relic.py | 23 +-- res_func/relic_res.py | 9 +- res_func/url.py | 3 + res_func/yatta/avatar.py | 25 ++-- 26 files changed, 422 insertions(+), 549 deletions(-) delete mode 100644 func/README.md create mode 100644 func/data.py delete mode 100644 models/monster.py diff --git a/fix_data.py b/fix_data.py index 7f60249..c5a5634 100644 --- a/fix_data.py +++ b/fix_data.py @@ -11,10 +11,10 @@ async def main(): text_map_data = await fetch_text_map() await fix_avatar_config(text_map_data) await fix_light_cone_config() - await fetch_relic_config(text_map_data) + await fetch_relic_config() await fix_set_image() await get_all_avatars() -if __name__ == '__main__': +if __name__ == "__main__": asyncio.run(main()) diff --git a/func/README.md b/func/README.md deleted file mode 100644 index 82bae62..0000000 --- a/func/README.md +++ /dev/null @@ -1,3 +0,0 @@ -# wiki 解析 - -通过米游社官方 wiki 解析部分数据 diff --git a/func/client.py b/func/client.py index fdf7efc..064a5be 100644 --- a/func/client.py +++ b/func/client.py @@ -1,19 +1,32 @@ -from httpx import AsyncClient +import asyncio +from httpx import AsyncClient, TimeoutException headers = { - 'authority': 'api-static.mihoyo.com', - 'accept': 'application/json, text/plain, */*', - 'accept-language': 'zh-CN,zh;q=0.9,zh-Hans;q=0.8,und;q=0.7,en;q=0.6,zh-Hant;q=0.5,ja;q=0.4', - 'dnt': '1', - 'origin': 'https://bbs.mihoyo.com', - 'referer': 'https://bbs.mihoyo.com/', - 'sec-ch-ua': '"Chromium";v="110", "Not A(Brand";v="24"', - 'sec-ch-ua-mobile': '?0', - 'sec-ch-ua-platform': '"Windows"', - 'sec-fetch-dest': 'empty', - 'sec-fetch-mode': 'cors', - 'sec-fetch-site': 'same-site', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36', + "authority": "api-static.mihoyo.com", + "accept": "application/json, text/plain, */*", + "accept-language": "zh-CN,zh;q=0.9,zh-Hans;q=0.8,und;q=0.7,en;q=0.6,zh-Hant;q=0.5,ja;q=0.4", + "dnt": "1", + "origin": "https://bbs.mihoyo.com", + "referer": "https://bbs.mihoyo.com/", + "sec-ch-ua": '"Chromium";v="110", "Not A(Brand";v="24"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Windows"', + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-site", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36", } client = AsyncClient(headers=headers, timeout=120.0) + + +def retry(func): + async def wrapper(*args, **kwargs): + for i in range(3): + try: + return await func(*args, **kwargs) + except TimeoutException: + print(f"重试 {func.__name__} {i + 1} 次") + await asyncio.sleep(1) + + return wrapper diff --git a/func/data.py b/func/data.py new file mode 100644 index 0000000..593885a --- /dev/null +++ b/func/data.py @@ -0,0 +1,104 @@ +from pathlib import Path +from typing import List, Dict + +import aiofiles +import ujson + +from models.avatar import YattaAvatar +from models.light_cone import YattaLightCone +from models.material import YattaMaterial +from models.relic import YattaRelic + +data_path = Path("data") +data_path.mkdir(exist_ok=True) + +avatars_path = data_path / "avatars.json" +all_avatars: List[YattaAvatar] = [] +all_avatars_map: Dict[int, YattaAvatar] = {} +all_avatars_name: Dict[str, YattaAvatar] = {} + +light_cones_path = data_path / "light_cones.json" +all_light_cones: List[YattaLightCone] = [] +all_light_cones_map: Dict[int, YattaLightCone] = {} +all_light_cones_name: Dict[str, YattaLightCone] = {} + +materials_path = data_path / "materials.json" +all_materials: List[YattaMaterial] = [] +all_materials_map: Dict[int, YattaMaterial] = {} +all_materials_name: Dict[str, YattaMaterial] = {} + +relics_path = data_path / "relics.json" +all_relics: List[YattaRelic] = [] + + +async def dump_avatars(): + data = [avatar.dict() for avatar in all_avatars] + data.sort(key=lambda x: x["id"]) + async with aiofiles.open(avatars_path, "w", encoding="utf-8") as f: + await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) + + +async def read_avatars(): + all_avatars.clear() + all_avatars_map.clear() + all_avatars_name.clear() + async with aiofiles.open(avatars_path, "r", encoding="utf-8") as f: + data = ujson.loads(await f.read()) + for avatar in data: + m = YattaAvatar(**avatar) + all_avatars.append(m) + all_avatars_map[m.id] = m + all_avatars_name[m.name] = m + + +async def dump_light_cones(): + data = [light_cone.dict() for light_cone in all_light_cones] + data.sort(key=lambda x: x["id"]) + async with aiofiles.open(light_cones_path, "w", encoding="utf-8") as f: + await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) + + +async def read_light_cones(): + async with aiofiles.open(light_cones_path, "r", encoding="utf-8") as f: + data = ujson.loads(await f.read()) + all_light_cones.clear() + all_light_cones_map.clear() + all_light_cones_name.clear() + for light_cone in data: + m = YattaLightCone(**light_cone) + all_light_cones.append(m) + all_light_cones_map[m.id] = m + all_light_cones_name[m.name] = m + + +async def dump_materials(): + data = [material.dict() for material in all_materials] + data.sort(key=lambda x: x["id"]) + async with aiofiles.open(materials_path, "w", encoding="utf-8") as f: + await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) + + +async def read_materials(): + async with aiofiles.open(materials_path, "r", encoding="utf-8") as f: + data = ujson.loads(await f.read()) + for material in data: + m = YattaMaterial(**material) + all_materials.append(m) + all_materials_map[m.id] = m + all_materials_name[m.name] = m + + +async def dump_relics(): + data = [relic.dict() for relic in all_relics] + data.sort(key=lambda x: x["id"]) + async with aiofiles.open(relics_path, "w", encoding="utf-8") as f: + await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) + + +async def read_relics(): + all_relics.clear() + async with aiofiles.open(relics_path, "r", encoding="utf-8") as f: + data = ujson.loads(await f.read()) + for relic in data: + m = YattaRelic(**relic) + all_relics.append(m) diff --git a/func/fetch_all.py b/func/fetch_all.py index 4a1d75e..192a21c 100644 --- a/func/fetch_all.py +++ b/func/fetch_all.py @@ -5,10 +5,10 @@ from func.url import list_url from models.wiki import Children -async def get_list(channel_id: str = '17') -> List[Children]: +async def get_list(channel_id: str = "17") -> List[Children]: params = { - 'app_sn': 'sr_wiki', - 'channel_id': channel_id, + "app_sn": "sr_wiki", + "channel_id": channel_id, } resp = await client.get(list_url, params=params) children = resp.json()["data"]["list"][0]["children"] diff --git a/func/fetch_avatars.py b/func/fetch_avatars.py index 09c766f..77082cb 100644 --- a/func/fetch_avatars.py +++ b/func/fetch_avatars.py @@ -1,33 +1,13 @@ -import asyncio -from pathlib import Path from typing import List, Dict -import aiofiles -import ujson -from httpx import TimeoutException from pydantic import ValidationError -from func.client import client +from func.client import client, retry +from func.data import all_avatars, all_avatars_map, all_avatars_name, dump_avatars from models.avatar import YattaAvatar from models.wiki import Content, Children from res_func.url import avatar_yatta_url -all_avatars: List[YattaAvatar] = [] -all_avatars_map: Dict[int, YattaAvatar] = {} -all_avatars_name: Dict[str, YattaAvatar] = {} - - -def retry(func): - async def wrapper(*args, **kwargs): - for i in range(3): - try: - return await func(*args, **kwargs) - except TimeoutException: - print(f"重试 {func.__name__} {i + 1} 次") - await asyncio.sleep(1) - - return wrapper - def fix_avatar_eidolons(values: Dict) -> Dict: if values.get("eidolons") is None: @@ -81,23 +61,4 @@ async def fetch_avatars(child: Children): if not avatar.icon.startswith("http"): avatar.icon = "" print("获取角色数据完成") - - -async def dump_avatars(path: Path): - data = [avatar.dict() for avatar in all_avatars] - data.sort(key=lambda x: x["id"]) - async with aiofiles.open(path, "w", encoding="utf-8") as f: - await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) - - -async def read_avatars(path: Path): - all_avatars.clear() - all_avatars_map.clear() - all_avatars_name.clear() - async with aiofiles.open(path, "r", encoding="utf-8") as f: - data = ujson.loads(await f.read()) - for avatar in data: - m = YattaAvatar(**avatar) - all_avatars.append(m) - all_avatars_map[m.id] = m - all_avatars_name[m.name] = m + await dump_avatars() diff --git a/func/fetch_light_cones.py b/func/fetch_light_cones.py index ffe38c9..05db160 100644 --- a/func/fetch_light_cones.py +++ b/func/fetch_light_cones.py @@ -1,119 +1,38 @@ -import asyncio -import re -from pathlib import Path -from typing import List, Dict +from typing import List -import aiofiles -import ujson -from bs4 import BeautifulSoup +from pydantic import ValidationError -from func.client import client -from func.fetch_materials import all_materials_map, all_materials_name -from func.url import info_url -from models.enums import Quality, Destiny -from models.light_cone import LightCone, LightConePromote, LightConeItem -from models.wiki import Children - -all_light_cones: List[LightCone] = [] -all_light_cones_name: Dict[str, LightCone] = {} +from func.client import client, retry +from func.data import all_light_cones, all_light_cones_name, dump_light_cones +from models.light_cone import YattaLightCone +from res_func.url import light_cone_yatta_url -async def fetch_light_cones(data: Children): - for content in data.list: - m_destiny = Destiny(re.findall(r'命途/(.*?)\\', content.ext)[0]) - m_quality = Quality(re.findall(r'星级/(.*?)\\', content.ext)[0]) - light_cone = LightCone( - id=content.content_id, - name=content.title, - desc=content.summary, - icon=content.icon, - big_pic="", - quality=m_quality, - destiny=m_destiny, - promote=[], - ) - all_light_cones.append(light_cone) - all_light_cones_name[light_cone.name] = light_cone +@retry +async def get_single_light_cone(url: str) -> None: + req = await client.get(url) + try: + light_cone = YattaLightCone(**(req.json()["data"])) + except Exception as e: + print(f"{url} 获取光锥数据失败") + raise e + all_light_cones.append(light_cone) + all_light_cones_name[light_cone.name] = light_cone -def parse_promote(light_cone: LightCone, soup: BeautifulSoup) -> None: - """解析光锥突破数据""" - mater = soup.find("div", {"data-part": "material"}) - trs = mater.find_all("tr") - required_levels = [20, 30, 40, 50, 60, 70] - max_level = [30, 40, 50, 60, 70, 80] - for i in range(0, min(len(trs), 6)): - promote = LightConePromote( - required_level=required_levels[i], - max_level=max_level[i], - items=[], - ) - materials = trs[i].find_all("li", {"class": "obc-tmpl__material-item"}) - for material in materials: - try: - mid = int(re.findall(r"content/(\d+)/detail", material.find("a").get("href"))[0]) - except AttributeError: - continue - name = material.find("p", {"class": "obc-tmpl__material-name"}).text - item = all_materials_map.get(mid) - if not item: - item = all_materials_name.get(name) - try: - count = int(material.find("span", {"class": "obc-tmpl__material-num"}).text) - except (AttributeError, ValueError): - count = 1 - if name == "信用点": - promote.coin = count - elif item: - promote.items.append( - LightConeItem( - item=item, - count=count, - ) - ) - else: - print(f"unknown material: {mid}: {name}") - light_cone.promote.append(promote) +@retry +async def get_all_light_cones() -> List[str]: + req = await client.get(light_cone_yatta_url) + return list(req.json()["data"]["items"].keys()) -async def fetch_info(light_cone: LightCone): - print(f"Fetch light_cone info: {light_cone.id}: {light_cone.name}") - params = { - 'app_sn': 'sr_wiki', - 'content_id': str(light_cone.id), - } - resp = await client.get(info_url, params=params) - data = resp.json()["data"]["content"]["contents"][0]["text"] - soup = BeautifulSoup(data, "lxml") - table = soup.find("table", {"class": "obc-tml-light-table--pc"}) - tr = table.find_all("tr")[-1] - td = tr.find_all("td")[-1] - light_cone.desc = td.get_text().strip() - pic_td = soup.find("td", {"class": "obc-tmpl-character__avatar"}) - light_cone.big_pic = pic_td.find("img").get("src") - parse_promote(light_cone, soup) - - -async def fetch_light_cones_infos(): - tasks = [] - for light_cone in all_light_cones: - tasks.append(fetch_info(light_cone)) - await asyncio.gather(*tasks) - - -async def dump_light_cones(path: Path): - data = [light_cone.dict() for light_cone in all_light_cones] - data.sort(key=lambda x: x["id"]) - async with aiofiles.open(path, "w", encoding="utf-8") as f: - await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) - - -async def read_light_cones(path: Path): - async with aiofiles.open(path, "r", encoding="utf-8") as f: - data = ujson.loads(await f.read()) - all_light_cones.clear() - all_light_cones_name.clear() - for light_cone in data: - m = LightCone(**light_cone) - all_light_cones.append(m) - all_light_cones_name[m.name] = m +async def fetch_light_cones(): + print("获取光锥数据") + light_cones = await get_all_light_cones() + for light_cone_id in light_cones: + try: + await get_single_light_cone(f"{light_cone_yatta_url}/{light_cone_id}") + except ValidationError: + print(f"{light_cone_yatta_url}/{light_cone_id} 获取光锥数据失败，数据格式异常") + print("获取光锥数据完成") + await dump_light_cones() diff --git a/func/fetch_materials.py b/func/fetch_materials.py index 8d435da..cb608e3 100644 --- a/func/fetch_materials.py +++ b/func/fetch_materials.py @@ -1,93 +1,50 @@ -import asyncio -import re -from pathlib import Path from typing import List, Dict -import aiofiles -import ujson -from bs4 import BeautifulSoup +from pydantic import ValidationError -from func.client import client -from func.url import info_url -from models.enums import Quality, MaterialType -from models.material import Material -from models.wiki import Children - -star_map = { - 1: Quality.One, - 2: Quality.Two, - 3: Quality.Three, - 4: Quality.Four, - 5: Quality.Five, -} -all_materials: List[Material] = [] -all_materials_map: Dict[int, Material] = {} -all_materials_name: Dict[str, Material] = {} +from func.client import client, retry +from func.data import ( + all_materials, + all_materials_map, + all_materials_name, + dump_materials, +) +from models.material import YattaMaterial +from res_func.url import material_yatta_url -async def fetch_materials(data: Children, default: str = "其他材料"): - for content in data.list: - quality: Quality = Quality.Three - if result := re.search(r"(\d+)星", content.ext): - quality: Quality = star_map[int(result[1])] +def fix_material(values: Dict) -> Dict: + if values.get("source") is None: + values["source"] = [] + return values + + +@retry +async def get_single_material(url: str) -> None: + req = await client.get(url) + try: + material = YattaMaterial(**fix_material(req.json()["data"])) + except Exception as e: + print(f"{url} 获取材料数据失败") + raise e + all_materials.append(material) + all_materials_map[material.id] = material + all_materials_name[material.name] = material + + +@retry +async def get_all_materials() -> List[str]: + req = await client.get(material_yatta_url) + return list(req.json()["data"]["items"].keys()) + + +async def fetch_materials(): + print("获取材料数据") + materials = await get_all_materials() + for material_id in materials: try: - m_type = re.findall(r'用途/(.*?)\\', content.ext)[0] - except IndexError: - m_type = default - if m_type == "角色晋阶": - m_type = "角色晋阶材料" - m_type = MaterialType(m_type) - material = Material( - id=content.content_id, - name=content.title, - desc=content.summary, - icon=content.icon, - quality=quality, - type=m_type, - ) - all_materials.append(material) - all_materials_map[material.id] = material - all_materials_name[material.name] = material - - -async def fetch_info(material: Material): - print(f"Fetch material info: {material.id}: {material.name}") - params = { - 'app_sn': 'sr_wiki', - 'content_id': str(material.id), - } - resp = await client.get(info_url, params=params) - data = resp.json()["data"]["content"]["contents"][0]["text"] - soup = BeautifulSoup(data, "lxml") - table = soup.find("table", {"class": "material-table--pc"}) - if result := re.search(r"(\d+)星", table.text): - material.quality = star_map[int(result[1])] - ps = table.find_all("p", {"style": "white-space: pre-wrap;"}) - text = "" - for p in ps: - text += f"{p.get_text()}\n" - material.desc = text.strip() - - -async def fetch_materials_infos(): - tasks = [] - for material in all_materials: - tasks.append(fetch_info(material)) - await asyncio.gather(*tasks) - - -async def dump_materials(path: Path): - data = [material.dict() for material in all_materials] - data.sort(key=lambda x: x["id"]) - async with aiofiles.open(path, "w", encoding="utf-8") as f: - await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) - - -async def read_materials(path: Path): - async with aiofiles.open(path, "r", encoding="utf-8") as f: - data = ujson.loads(await f.read()) - for material in data: - m = Material(**material) - all_materials.append(m) - all_materials_map[m.id] = m - all_materials_name[m.name] = m + await get_single_material(f"{material_yatta_url}/{material_id}") + except ValidationError: + print(f"{material_yatta_url}/{material_id} 获取材料数据失败，数据格式异常") + print("获取材料数据完成") + await dump_materials() diff --git a/func/fetch_relics.py b/func/fetch_relics.py index 2261e54..5549595 100644 --- a/func/fetch_relics.py +++ b/func/fetch_relics.py @@ -1,62 +1,28 @@ -import asyncio -from pathlib import Path -from typing import List +from typing import List, Dict -import aiofiles -import ujson -from bs4 import BeautifulSoup +from pydantic import ValidationError -from func.client import client -from func.url import info_url -from models.relic import Relic -from models.wiki import Children - -all_relics: List[Relic] = [] +from func.client import client, retry +from func.data import all_relics, dump_relics +from models.relic import YattaRelic +from res_func.url import relic_yatta_url -async def fetch_relics(data: Children): - for content in data.list: - relic = Relic( - id=content.content_id, - bbs_id=content.content_id, - name=content.title, - icon=content.icon, - affect="", - image_list=[], - ) - all_relics.append(relic) +@retry +async def get_all_relics() -> List[Dict]: + req = await client.get(relic_yatta_url) + return list(req.json()["data"]["items"].values()) -async def fetch_info(relic: Relic): - print(f"Fetch relic info: {relic.id}: {relic.name}") - params = { - 'app_sn': 'sr_wiki', - 'content_id': str(relic.id), - } - resp = await client.get(info_url, params=params) - data = resp.json()["data"]["content"]["contents"][0]["text"] - soup = BeautifulSoup(data, "lxml") - relic.affect = soup.find("div", {"class": "obc-tmpl-relic__story"}).get_text().strip() - - -async def fetch_relics_infos(): - tasks = [] - for relic in all_relics: - tasks.append(fetch_info(relic)) - await asyncio.gather(*tasks) - - -async def dump_relics(path: Path): - data = [relic.dict() for relic in all_relics] - data.sort(key=lambda x: x["id"]) - async with aiofiles.open(path, "w", encoding="utf-8") as f: - await f.write(ujson.dumps(data, indent=4, ensure_ascii=False)) - - -async def read_relics(path: Path): - all_relics.clear() - async with aiofiles.open(path, "r", encoding="utf-8") as f: - data = ujson.loads(await f.read()) - for relic in data: - m = Relic(**relic) - all_relics.append(m) +async def fetch_relics(): + print("获取遗器数据") + relics = await get_all_relics() + for relic in relics: + try: + relic_ = YattaRelic(**relic) + all_relics.append(relic_) + except ValidationError as e: + raise e + print(f"{relic} 解析遗器数据失败，数据格式异常") + print("获取遗器数据完成") + await dump_relics() diff --git a/func/url.py b/func/url.py index f854764..97b47df 100644 --- a/func/url.py +++ b/func/url.py @@ -1,3 +1,2 @@ base_url = "https://api-static.mihoyo.com/common/blackboard/sr_wiki/v1" list_url = f"{base_url}/home/content/list" -info_url = f"{base_url}/content/info" diff --git a/main.py b/main.py index 371fa8a..b4af114 100644 --- a/main.py +++ b/main.py @@ -1,78 +1,29 @@ import asyncio -from pathlib import Path + from func.fetch_all import get_list -from func.fetch_avatars import fetch_avatars, dump_avatars, read_avatars -from func.fetch_light_cones import fetch_light_cones, fetch_light_cones_infos, dump_light_cones, read_light_cones -from func.fetch_materials import fetch_materials, fetch_materials_infos, dump_materials, read_materials -from func.fetch_relics import fetch_relics, fetch_relics_infos, dump_relics, read_relics +from func.fetch_avatars import fetch_avatars +from func.fetch_light_cones import fetch_light_cones +from func.fetch_materials import fetch_materials +from func.fetch_relics import fetch_relics from func.fetch_src import move_files -data_path = Path("data") -data_path.mkdir(exist_ok=True) - -async def wiki( - override_materials: bool = True, - override_avatars: bool = True, - override_light_cones: bool = True, - override_relics: bool = True, -): +async def wiki(): main_data = await get_list() - if override_materials: - await fetch_materials(main_data[4]) - await fetch_materials(main_data[5]) - await fetch_materials(main_data[12], "消耗品") - await fetch_materials(main_data[10], "任务道具") - await fetch_materials(main_data[13], "贵重物") - await fetch_materials_infos() - await dump_materials(data_path / "materials.json") - else: - await read_materials(data_path / "materials.json") - if override_avatars: - await fetch_avatars(main_data[0]) - await dump_avatars(data_path / "avatars.json") - else: - await read_avatars(data_path / "avatars.json") - if override_light_cones: - await fetch_light_cones(main_data[1]) - await fetch_light_cones_infos() - await dump_light_cones(data_path / "light_cones.json") - else: - await read_light_cones(data_path / "light_cones.json") - if override_relics: - await fetch_relics(main_data[2]) - await fetch_relics_infos() - await dump_relics(data_path / "relics.json") - else: - await read_relics(data_path / "relics.json") + await fetch_avatars(main_data[0]) + await fetch_light_cones() + await fetch_materials() + await fetch_relics() async def bbs_photos(): await move_files() -async def main( - override_materials: bool = True, - override_avatars: bool = True, - override_light_cones: bool = True, - override_relics: bool = True, - override_bbs_photos: bool = True, -): - await wiki(override_materials, override_avatars, override_light_cones, override_relics) - if override_bbs_photos: - await bbs_photos() +async def main(): + await wiki() + await bbs_photos() -if __name__ == '__main__': - override_material_ = True - override_avatar_ = True - override_light_cone_ = True - override_relic_ = True - override_bbs_photo_ = True - asyncio.run(main( - override_material_, - override_avatar_, - override_light_cone_, - override_relic_, - override_bbs_photo_, - )) +if __name__ == "__main__": + asyncio.run(main()) diff --git a/models/avatar.py b/models/avatar.py index 47f536a..665ae60 100644 --- a/models/avatar.py +++ b/models/avatar.py @@ -59,10 +59,10 @@ class YattaAvatar(BaseModel): @property def destiny(self) -> Destiny: - """ 命途 """ + """命途""" return Destiny(self.types.pathType.name) @property def element(self) -> Element: - """ 属性 """ + """属性""" return Element(self.types.combatType.name) diff --git a/models/light_cone.py b/models/light_cone.py index 21a6e24..21e013d 100644 --- a/models/light_cone.py +++ b/models/light_cone.py @@ -1,51 +1,29 @@ # 光锥 -from typing import List - from pydantic import BaseModel -from .enums import Quality, Destiny -from .material import Material + +class YattaLightConePath(BaseModel): + id: str + name: str -class LightConeItem(BaseModel): - item: Material - """物品""" - count: int - """数量""" +class YattaLightConeTypes(BaseModel): + pathType: YattaLightConePath -class LightConePromote(BaseModel): - required_level: int - """突破所需等级""" - promote_level: int = 0 - """突破等级""" - max_level: int - """解锁的等级上限""" - - coin: int = 0 - """信用点""" - items: List[LightConeItem] - """突破所需材料""" - - -class LightCone(BaseModel): +class YattaLightCone(BaseModel): id: int """"光锥ID""" name: str """名称""" - desc: str + description: str """描述""" - icon: str + icon: str = "" """图标""" - big_pic: str + big_pic: str = "" """大图""" - quality: Quality + rank: int """稀有度""" - destiny: Destiny + types: YattaLightConeTypes """命途""" - promote: List[LightConePromote] - """晋阶信息""" - - @property - def rarity(self) -> int: - return 5 - list(Quality).index(self.quality) + route: str diff --git a/models/material.py b/models/material.py index 88ab22f..bd8a884 100644 --- a/models/material.py +++ b/models/material.py @@ -1,19 +1,35 @@ # 材料 +from typing import List, Optional + from pydantic import BaseModel -from .enums import Quality, MaterialType + +class MaterialSource(BaseModel): + description: str -class Material(BaseModel): +class MaterialMType(BaseModel): + id: int + name: str + + +class YattaMaterial(BaseModel): id: int """材料ID""" name: str """名称""" - desc: str - """介绍""" - icon: str - """图标""" - quality: Quality + description: str + """描述""" + story: str + """故事""" + rank: int """稀有度""" - type: MaterialType + source: List[MaterialSource] + """来源""" + type: Optional[MaterialMType] = None """类型""" + route: str + + @property + def icon(self) -> str: + return f"https://api.yatta.top/hsr/assets/UI/item/{self.id}.png" diff --git a/models/monster.py b/models/monster.py deleted file mode 100644 index 474f3c7..0000000 --- a/models/monster.py +++ /dev/null @@ -1,25 +0,0 @@ -# 敌对物种 -from pydantic import BaseModel - -from .enums import MonsterType, Area - - -class Monster(BaseModel): - id: int - """怪物ID""" - name: str - """名称""" - desc: str - """介绍""" - icon: str - """图标""" - big_pic: str - """大图""" - type: MonsterType - """种类""" - area: Area - """地区""" - resistance: str - """抗性""" - find_area: str - """发现地点""" diff --git a/models/relic.py b/models/relic.py index 7de7f33..8dfce8c 100644 --- a/models/relic.py +++ b/models/relic.py @@ -4,16 +4,13 @@ from typing import List from pydantic import BaseModel -class Relic(BaseModel): +class YattaRelic(BaseModel): id: int """遗器套装ID""" - bbs_id: int - """WIKI ID""" name: str """套装名称""" icon: str """套装图标""" - affect: str - """套装效果""" - image_list: List[str] + image_list: List[str] = [] """套装子图""" + route: str diff --git a/modules/apihelper/client/base/httpxrequest.py b/modules/apihelper/client/base/httpxrequest.py index 022c1d4..7524a7d 100644 --- a/modules/apihelper/client/base/httpxrequest.py +++ b/modules/apihelper/client/base/httpxrequest.py @@ -16,7 +16,9 @@ timeout = httpx.Timeout( class HTTPXRequest(AbstractAsyncContextManager): def __init__(self, *args, headers=None, **kwargs): - self._client = httpx.AsyncClient(headers=headers, timeout=timeout, *args, **kwargs) + self._client = httpx.AsyncClient( + headers=headers, timeout=timeout, *args, **kwargs + ) async def __aenter__(self): try: @@ -27,8 +29,10 @@ class HTTPXRequest(AbstractAsyncContextManager): raise exc async def __aexit__( - self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], - exc_tb: Optional[TracebackType] + self, + exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[TracebackType], ) -> None: await self.initialize() diff --git a/modules/apihelper/client/base/hyperionrequest.py b/modules/apihelper/client/base/hyperionrequest.py index 1495a54..d38aa96 100644 --- a/modules/apihelper/client/base/hyperionrequest.py +++ b/modules/apihelper/client/base/hyperionrequest.py @@ -12,16 +12,25 @@ __all__ = ("HyperionRequest",) class HyperionRequest(HTTPXRequest): async def get( - self, url: str, *args, de_json: bool = True, re_json_data: bool = False, **kwargs + self, + url: str, + *args, + de_json: bool = True, + re_json_data: bool = False, + **kwargs, ) -> Union[POST_DATA, JSON_DATA, Response]: try: response = await self._client.get(url=url, *args, **kwargs) except httpx.TimeoutException as err: raise APIHelperTimedOut from err except httpx.HTTPError as exc: - raise NetworkException(f"Unknown error in HTTP implementation: {repr(exc)}") from exc + raise NetworkException( + f"Unknown error in HTTP implementation: {repr(exc)}" + ) from exc if response.is_error: - raise ResponseException(message=f"response error in status code: {response.status_code}") + raise ResponseException( + message=f"response error in status code: {response.status_code}" + ) if not de_json: return response json_data = response.json() @@ -32,23 +41,34 @@ class HyperionRequest(HTTPXRequest): return json_data if return_code != 0: if message is None: - raise ResponseException(message=f"response error in return code: {return_code}") + raise ResponseException( + message=f"response error in return code: {return_code}" + ) raise ResponseException(response=json_data) if not re_json_data and data is not None: return data return json_data async def post( - self, url: str, *args, de_json: bool = True, re_json_data: bool = False, **kwargs + self, + url: str, + *args, + de_json: bool = True, + re_json_data: bool = False, + **kwargs, ) -> Union[POST_DATA, JSON_DATA, Response]: try: response = await self._client.post(url=url, *args, **kwargs) except httpx.TimeoutException as err: raise APIHelperTimedOut from err except httpx.HTTPError as exc: - raise NetworkException(f"Unknown error in HTTP implementation: {repr(exc)}") from exc + raise NetworkException( + f"Unknown error in HTTP implementation: {repr(exc)}" + ) from exc if response.is_error: - raise ResponseException(message=f"response error in status code: {response.status_code}") + raise ResponseException( + message=f"response error in status code: {response.status_code}" + ) if not de_json: return response json_data = response.json() @@ -59,7 +79,9 @@ class HyperionRequest(HTTPXRequest): return json_data if return_code != 0: if message is None: - raise ResponseException(message=f"response error in return code: {return_code}") + raise ResponseException( + message=f"response error in return code: {return_code}" + ) raise ResponseException(response=json_data) if not re_json_data and data is not None: return data diff --git a/modules/apihelper/client/components/hyperion.py b/modules/apihelper/client/components/hyperion.py index 3c8d7e6..8b0ecb6 100644 --- a/modules/apihelper/client/components/hyperion.py +++ b/modules/apihelper/client/components/hyperion.py @@ -17,9 +17,13 @@ class Hyperion: """ POST_FULL_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFull" - POST_FULL_IN_COLLECTION_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFullInCollection" + POST_FULL_IN_COLLECTION_URL = ( + "https://bbs-api.miyoushe.com/post/wapi/getPostFullInCollection" + ) GET_NEW_LIST_URL = "https://bbs-api.miyoushe.com/post/wapi/getNewsList" - GET_OFFICIAL_RECOMMENDED_POSTS_URL = "https://bbs-api.miyoushe.com/post/wapi/getOfficialRecommendedPosts" + GET_OFFICIAL_RECOMMENDED_POSTS_URL = ( + "https://bbs-api.miyoushe.com/post/wapi/getOfficialRecommendedPosts" + ) USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " @@ -39,7 +43,9 @@ class Hyperion: # https://m.miyoushe.com/ys/#/article/32497914 :return: post_id """ - rgx = re.compile(r"(?:bbs|www\.)?(?:miyoushe|mihoyo)\.com/[^.]+/article/(?P\d+)") + rgx = re.compile( + r"(?:bbs|www\.)?(?:miyoushe|mihoyo)\.com/[^.]+/article/(?P\d+)" + ) matches = rgx.search(text) if matches is None: return -1 @@ -56,7 +62,9 @@ class Hyperion: return {"User-Agent": self.USER_AGENT, "Referer": referer} @staticmethod - def get_list_url_params(forum_id: int, is_good: bool = False, is_hot: bool = False, page_size: int = 20) -> dict: + def get_list_url_params( + forum_id: int, is_good: bool = False, is_hot: bool = False, page_size: int = 20 + ) -> dict: return { "forum_id": forum_id, "gids": 2, @@ -68,7 +76,11 @@ class Hyperion: @staticmethod def get_images_params( - resize: int = 600, quality: int = 80, auto_orient: int = 0, interlace: int = 1, images_format: str = "jpg" + resize: int = 600, + quality: int = 80, + auto_orient: int = 0, + interlace: int = 1, + images_format: str = "jpg", ): """ image/resize,s_600/quality,q_80/auto-orient,0/interlace,1/format,jpg @@ -87,12 +99,22 @@ class Hyperion: async def get_official_recommended_posts(self, gids: int) -> JSON_DATA: params = {"gids": gids} - response = await self.client.get(url=self.GET_OFFICIAL_RECOMMENDED_POSTS_URL, params=params) + response = await self.client.get( + url=self.GET_OFFICIAL_RECOMMENDED_POSTS_URL, params=params + ) return response - async def get_post_full_in_collection(self, collection_id: int, gids: int = 2, order_type=1) -> JSON_DATA: - params = {"collection_id": collection_id, "gids": gids, "order_type": order_type} - response = await self.client.get(url=self.POST_FULL_IN_COLLECTION_URL, params=params) + async def get_post_full_in_collection( + self, collection_id: int, gids: int = 2, order_type=1 + ) -> JSON_DATA: + params = { + "collection_id": collection_id, + "gids": gids, + "order_type": order_type, + } + response = await self.client.get( + url=self.POST_FULL_IN_COLLECTION_URL, params=params + ) return response async def get_post_info(self, gids: int, post_id: int, read: int = 1) -> PostInfo: @@ -100,7 +122,9 @@ class Hyperion: response = await self.client.get(self.POST_FULL_URL, params=params) return PostInfo.paste_data(response) - async def get_images_by_post_id(self, gids: int, post_id: int) -> List[ArtworkImage]: + async def get_images_by_post_id( + self, gids: int, post_id: int + ) -> List[ArtworkImage]: post_info = await self.get_post_info(gids, post_id) art_list = [] task_list = [ @@ -118,15 +142,23 @@ class Hyperion: art_list.sort(key=take_page) return art_list - async def download_image(self, art_id: int, url: str, page: int = 0) -> ArtworkImage: + async def download_image( + self, art_id: int, url: str, page: int = 0 + ) -> ArtworkImage: filename = os.path.basename(url) _, file_extension = os.path.splitext(filename) is_image = bool(file_extension in ".jpg" or file_extension in ".png") response = await self.client.get( - url, params=self.get_images_params(resize=2000) if is_image else None, de_json=False + url, + params=self.get_images_params(resize=2000) if is_image else None, + de_json=False, ) return ArtworkImage( - art_id=art_id, page=page, file_name=filename, file_extension=url.split(".")[-1], data=response.content + art_id=art_id, + page=page, + file_name=filename, + file_extension=url.split(".")[-1], + data=response.content, ) async def get_new_list(self, gids: int, type_id: int, page_size: int = 20): diff --git a/modules/apihelper/error.py b/modules/apihelper/error.py index 42c29cf..a388f44 100644 --- a/modules/apihelper/error.py +++ b/modules/apihelper/error.py @@ -17,7 +17,11 @@ class ResponseException(APIHelperException): code: int = 0 message: str = "" - def __init__(self, response: Optional[Mapping[str, Any]] = None, message: Optional[str] = None) -> None: + def __init__( + self, + response: Optional[Mapping[str, Any]] = None, + message: Optional[str] = None, + ) -> None: if response is None: self.message = message _message = message diff --git a/res_func/avatar.py b/res_func/avatar.py index 1568fee..3f0920c 100644 --- a/res_func/avatar.py +++ b/res_func/avatar.py @@ -7,7 +7,7 @@ import aiofiles import ujson from bs4 import BeautifulSoup, Tag -from func.fetch_avatars import read_avatars, all_avatars_name, dump_avatars, all_avatars_map +from func.data import all_avatars_map, all_avatars_name, read_avatars, dump_avatars from models.avatar_config import AvatarConfig, AvatarIcon from .client import client from .url import avatar_config, text_map, base_station_url, avatar_url @@ -48,8 +48,14 @@ async def parse_station(datas, name: str, tag: Tag, cid: int): third_pic = get_third_pic() text = soup.find("div", {"class": "a6678 a4af5"}).get("style") - four_pic = f'{base_station_url}{text[text.find("(") + 2:text.find(")") - 1]}' if text else "" - first_pic = f'{base_station_url}{soup.find("img", {"class": "ac39b a6602"}).get("src")}' + four_pic = ( + f'{base_station_url}{text[text.find("(") + 2:text.find(")") - 1]}' + if text + else "" + ) + first_pic = ( + f'{base_station_url}{soup.find("img", {"class": "ac39b a6602"}).get("src")}' + ) datas.append( AvatarIcon( id=cid, @@ -67,10 +73,7 @@ async def dump_icons(path: Path, datas: List[AvatarIcon]): async def fetch_station_ktz(tasks, datas, player_avatars: List[Tag]): - data_map = { - "开拓者·毁灭": (8001, 8002), - "开拓者·存护": (8003, 8004) - } + data_map = {"开拓者·毁灭": (8001, 8002), "开拓者·存护": (8003, 8004)} idx = 0 for key, value in data_map.items(): tasks.append(parse_station(datas, key, player_avatars[idx], value[0])) @@ -92,7 +95,11 @@ async def fetch_station(configs_map: Dict[str, AvatarConfig]) -> List[AvatarIcon player_avatars.append(avatar) continue avatar_model = configs_map.get(name) - tasks.append(parse_station(datas, name, avatar, avatar_model.AvatarID if avatar_model else None)) + tasks.append( + parse_station( + datas, name, avatar, avatar_model.AvatarID if avatar_model else None + ) + ) await fetch_station_ktz(tasks, datas, player_avatars) await asyncio.gather(*tasks) return datas @@ -110,8 +117,8 @@ async def fix_avatar_config(text_map_data: Dict[str, str]): configs_map: Dict[str, AvatarConfig] = {config.name: config for config in configs} print(f"读取到原始数据：{list(configs_map.keys())}") data_path = Path("data") - await read_avatars(data_path / "avatars.json") + await read_avatars() await fix_avatar_config_ktz() icons = await fetch_station(configs_map) await dump_icons(data_path / "avatar_icons.json", icons) - await dump_avatars(data_path / "avatars.json") + await dump_avatars() diff --git a/res_func/light_cone.py b/res_func/light_cone.py index 49ff368..dba673f 100644 --- a/res_func/light_cone.py +++ b/res_func/light_cone.py @@ -6,7 +6,7 @@ import aiofiles import ujson from bs4 import BeautifulSoup, Tag -from func.fetch_light_cones import read_light_cones, all_light_cones_name, dump_light_cones +from func.data import all_light_cones_map, read_light_cones, dump_light_cones from models.light_cone_config import LightConeIcon from .client import client from .url import base_station_url, light_cone_url @@ -18,6 +18,11 @@ async def parse_station(icon: LightConeIcon, tag: Tag): first_pic = f'{base_station_url}{soup.find("img", {"class": "standard-icon a6602"}).get("src")}' second_pic = f'{base_station_url}{soup.find("img", {"class": "a2b16 mobile-only-elem ab8c3"}).get("src")}' icon.icon = [first_pic, second_pic] + if light_cone_model := all_light_cones_map.get(icon.id): + light_cone_model.icon = first_pic + light_cone_model.big_pic = second_pic + else: + print(f"yatta 未找到光锥数据 {icon.name} ，修复 图标 失败") async def dump_icons(path: Path, datas: List[LightConeIcon]): @@ -42,15 +47,7 @@ async def fetch_station() -> List[LightConeIcon]: if "lightcone/" not in url: continue nid = int(url.split("/")[-1]) - if light_cone_model := all_light_cones_name.get(name): - light_cone_model.id = nid - else: - print(f"wiki 未找到光锥数据 {name} ，修复 id 失败") - icon = LightConeIcon( - id=nid, - name=name, - icon=[] - ) + icon = LightConeIcon(id=nid, name=name, icon=[]) datas.append(icon) tasks.append(parse_station(icon, light_cone)) await asyncio.gather(*tasks) @@ -59,7 +56,7 @@ async def fetch_station() -> List[LightConeIcon]: async def fix_light_cone_config(): data_path = Path("data") - await read_light_cones(data_path / "light_cones.json") + await read_light_cones() icons = await fetch_station() await dump_icons(data_path / "light_cone_icons.json", icons) - await dump_light_cones(data_path / "light_cones.json") + await dump_light_cones() diff --git a/res_func/relic.py b/res_func/relic.py index cd27884..07c85cc 100644 --- a/res_func/relic.py +++ b/res_func/relic.py @@ -4,11 +4,10 @@ from typing import List, Dict import aiofiles import ujson -from func.fetch_relics import read_relics, dump_relics, all_relics from models.enums import RelicAffix, RelicPosition from models.relic_affix import RelicAffixAll, SingleRelicAffix from res_func.client import client -from res_func.url import relic_config, relic_main_affix_config, relic_sub_affix_config, relic_set_config +from res_func.url import relic_config, relic_main_affix_config, relic_sub_affix_config final_datas: List[RelicAffixAll] = [] final_datas_map: Dict[str, RelicAffixAll] = {} @@ -78,23 +77,6 @@ async def fetch_sub_affix(): print("遗器副词条配置获取完毕") -async def fix_set_id(text_map_data: Dict[str, str]): - print("开始修复遗器套装ID") - set_req = await client.get(relic_set_config) - set_data = set_req.json() - set_data_map: Dict[str, int] = {} - for key, value in set_data.items(): - set_data_map[text_map_data[str(value["SetName"]["Hash"])]] = value["SetID"] - await read_relics(Path("data") / "relics.json") - for relic_set in all_relics: - if set_id := set_data_map.get(relic_set.name): - relic_set.id = set_id - else: - print(f"套装{relic_set.name}没有找到对应的ID") - await dump_relics(Path("data") / "relics.json") - print("遗器套装ID修复完毕") - - async def dump_relic_config(path: Path): final_data = [data.dict() for data in final_datas] final_data.sort(key=lambda x: x["id"]) @@ -102,10 +84,9 @@ async def dump_relic_config(path: Path): await f.write(ujson.dumps(final_data, indent=4, ensure_ascii=False)) -async def fetch_relic_config(text_map_data: Dict[str, str]): +async def fetch_relic_config(): await fetch_all_relic() await fetch_main_affix() await fetch_sub_affix() data_path = Path("data") await dump_relic_config(data_path / "relic_config.json") - await fix_set_id(text_map_data) diff --git a/res_func/relic_res.py b/res_func/relic_res.py index 0294803..e0a3206 100644 --- a/res_func/relic_res.py +++ b/res_func/relic_res.py @@ -1,18 +1,15 @@ -from pathlib import Path from typing import Dict, Tuple, List from bs4 import BeautifulSoup -from func.fetch_relics import all_relics, read_relics, dump_relics +from func.data import read_relics, all_relics, dump_relics from res_func.client import client from res_func.url import relic_url, base_station_url -relics_path = Path("data") / "relics.json" - async def fix_set_image(): print("开始修复遗器套装图片") - await read_relics(relics_path) + await read_relics() req = await client.get(relic_url) soup = BeautifulSoup(req.text, "lxml") divs = soup.find_all("a", {"class": "aff5a"}) @@ -36,5 +33,5 @@ async def fix_set_image(): relic.image_list = data_map[relic.id][1] else: print(f"套装 {relic.id} 没有找到对应的图片") - await dump_relics(relics_path) + await dump_relics() print("遗器套装图片修复完毕") diff --git a/res_func/url.py b/res_func/url.py index 3b52c61..31f4db9 100644 --- a/res_func/url.py +++ b/res_func/url.py @@ -16,3 +16,6 @@ relic_url = f"{base_station_url}/cn/relics" base_yatta_url = "https://api.yatta.top" avatar_yatta_url = f"{base_yatta_url}/hsr/v2/cn/avatar" avatar_skill_url = f"{base_yatta_url}/hsr/assets/UI/skill/" +light_cone_yatta_url = f"{base_yatta_url}/hsr/v2/cn/equipment" +material_yatta_url = f"{base_yatta_url}/hsr/v2/cn/item" +relic_yatta_url = f"{base_yatta_url}/hsr/v2/cn/relic" diff --git a/res_func/yatta/avatar.py b/res_func/yatta/avatar.py index 03d4a9a..f7de675 100644 --- a/res_func/yatta/avatar.py +++ b/res_func/yatta/avatar.py @@ -6,7 +6,8 @@ from typing import List import aiofiles import ujson -from func.fetch_avatars import all_avatars +from func.client import retry +from func.data import all_avatars from models.avatar import YattaAvatar from res_func.client import client from res_func.url import avatar_skill_url @@ -17,18 +18,6 @@ avatars_skills_path = Path("data/skill") avatars_skills_path.mkdir(exist_ok=True, parents=True) -def retry(func): - async def wrapper(*args, **kwargs): - for i in range(3): - try: - return await func(*args, **kwargs) - except Exception: - print(f"重试 {func.__name__} {i + 1} 次") - await asyncio.sleep(1) - - return wrapper - - @retry async def get_single_avatar_skill_icon(url: str, real_path: str) -> None: req = await client.get(url) @@ -51,7 +40,9 @@ async def get_single_avatar_skill_icon(url: str, real_path: str) -> None: async def dump_icons(): final_data = dict(sorted(avatar_data.items(), key=lambda x: x[0])) - async with aiofiles.open("data/avatar_eidolon_icons.json", "w", encoding="utf-8") as f: + async with aiofiles.open( + "data/avatar_eidolon_icons.json", "w", encoding="utf-8" + ) as f: await f.write(ujson.dumps(final_data, indent=4, ensure_ascii=False)) @@ -78,11 +69,13 @@ async def get_all_avatars_skills_icons(avatars: List[YattaAvatar]): tasks.append( get_single_avatar_skill_icon( f"{avatar_skill_url}SkillIcon_{avatar.id}_{remote_path[i]}.png", - f"{avatar.id}_{local_path[i]}.png" + f"{avatar.id}_{local_path[i]}.png", ) ) await asyncio.gather(*tasks) tasks.clear() datas = [file.name.split(".")[0] for file in avatars_skills_path.glob("*")] - async with aiofiles.open(avatars_skills_path / "info.json", "w", encoding="utf-8") as f: + async with aiofiles.open( + avatars_skills_path / "info.json", "w", encoding="utf-8" + ) as f: await f.write(json.dumps(datas, indent=4, ensure_ascii=False))