♻️ Use yatta parse avatars

func/fetch_all.py

@@ -2,7 +2,6 @@ from typing import List
 
 from func.client import client
 from func.url import list_url
-
 from models.wiki import Children
 
 

func/fetch_avatars.py

@@ -1,119 +1,86 @@
 import asyncio
-import re
 from pathlib import Path
 from typing import List, Dict
 
 import aiofiles
 import ujson
-from bs4 import BeautifulSoup
+from httpx import TimeoutException
+from pydantic import ValidationError
 
 from func.client import client
-from func.url import info_url
-from func.fetch_materials import all_materials_map, all_materials_name
-from models.enums import Quality, Element, Destiny
-from models.avatar import Avatar, AvatarInfo, AvatarSoul, AvatarItem, AvatarPromote
-from models.wiki import Children
+from models.avatar import YattaAvatar
+from models.wiki import Content, Children
+from res_func.url import avatar_yatta_url
 
-all_avatars: List[Avatar] = []
-all_avatars_map: Dict[int, Avatar] = {}
-all_avatars_name: Dict[str, Avatar] = {}
+all_avatars: List[YattaAvatar] = []
+all_avatars_map: Dict[int, YattaAvatar] = {}
+all_avatars_name: Dict[str, YattaAvatar] = {}
 
 
-async def fetch_avatars(data: Children):
-    for content in data.list:
+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:
+        values["eidolons"] = []
+    else:
+        eidolons = []
+        for eidolon in values["eidolons"].values():
+            eidolons.append(eidolon)
+        values["eidolons"] = eidolons
+    return values
+
+
+@retry
+async def get_single_avatar(url: str) -> None:
+    req = await client.get(url)
+    try:
+        avatar = YattaAvatar(**fix_avatar_eidolons(req.json()["data"]))
+    except Exception as e:
+        print(f"{url} 获取角色数据失败")
+        raise e
+    all_avatars.append(avatar)
+    all_avatars_map[avatar.id] = avatar
+    all_avatars_name[avatar.name] = avatar
+
+
+@retry
+async def get_all_avatar() -> List[str]:
+    req = await client.get(avatar_yatta_url)
+    return list(req.json()["data"]["items"].keys())
+
+
+async def fix_avatar_icon(content: Content):
+    avatar = all_avatars_name.get(content.title)
+    if not avatar:
+        return
+    avatar.icon = content.icon
+
+
+async def fetch_avatars(child: Children):
+    print("获取角色数据")
+    avatars = await get_all_avatar()
+    for avatar_id in avatars:
         try:
-            m_element = Element(re.findall(r'属性/(.*?)\\', content.ext)[0])
-            m_destiny = Destiny(re.findall(r'命途/(.*?)\\', content.ext)[0])
-            m_quality = Quality(re.findall(r'星级/(.*?)\\', content.ext)[0])
-        except IndexError:
-            continue
-        avatar = Avatar(
-            id=content.content_id,
-            name=content.title,
-            icon=content.icon,
-            quality=m_quality,
-            element=m_element,
-            destiny=m_destiny,
-            information=AvatarInfo(),
-            promote=[],
-            soul=[],
-        )
-        all_avatars.append(avatar)
-        all_avatars_map[avatar.id] = avatar
-        all_avatars_name[avatar.name] = avatar
-
-
-def parse_promote(avatar: Avatar, soup: BeautifulSoup) -> None:
-    """解析角色突破数据"""
-    lis = soup.find_all("li", {"class": "obc-tmpl__switch-item"})
-    required_levels = [0, 20, 30, 40, 50, 60, 70, 80]
-    max_level = [0, 30, 40, 50, 60, 70, 80, 90]
-    for i in range(1, 8):
-        promote = AvatarPromote(
-            required_level=required_levels[i],
-            max_level=max_level[i],
-            items=[],
-        )
-        materials = lis[i].find_all("li", {"data-target": "breach.attr.material"})
-        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("span", {"class": "obc-tmpl__icon-text"}).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__icon-num"}).text.replace("*", ""))
-            except AttributeError:
-                count = 1
-            if name == "信用点":
-                promote.coin = count
-            elif item:
-                promote.items.append(
-                    AvatarItem(
-                        item=item,
-                        count=count,
-                    )
-                )
-            else:
-                print(f"unknown material: {mid}: {name}")
-        avatar.promote.append(promote)
-
-
-async def fetch_info(avatar: Avatar):
-    print(f"Fetch avatar info: {avatar.id}: {avatar.name}")
-    params = {
-        'app_sn': 'sr_wiki',
-        'content_id': str(avatar.id),
-    }
-    resp = await client.get(info_url, params=params)
-    data = resp.json()["data"]["content"]["contents"][0]["text"]
-    soup = BeautifulSoup(data, "lxml")
-    items = soup.find_all("div", {"class": "obc-tmp-character__item"})
-    avatar.information.faction = items[2].find("div", {"class": "obc-tmp-character__value"}).text
-    avatar.information.occupation = items[3].find("div", {"class": "obc-tmp-character__value"}).text
-    parse_promote(avatar, soup)
-    # 星魂
-    table = soup.find("div", {"style": "order: 4;"})
-    trs = table.find_all("tr")[1:]
-    for tr in trs:
-        ps = tr.find_all("p")
-        desc = ps[2].text.strip() if len(ps) > 2 else ps[1].text.strip()
-        avatar.soul.append(
-            AvatarSoul(
-                name=ps[0].text.strip(),
-                desc=desc,
-            )
-        )
-
-
-async def fetch_avatars_infos():
-    tasks = []
+            await get_single_avatar(f"{avatar_yatta_url}/{avatar_id}")
+        except ValidationError:
+            print(f"{avatar_yatta_url}/{avatar_id} 获取角色数据失败，角色格式异常")
+    print("修复角色图标")
+    for content in child.list:
+        await fix_avatar_icon(content)
     for avatar in all_avatars:
-        tasks.append(fetch_info(avatar))
-    await asyncio.gather(*tasks)
+        if not avatar.icon.startswith("http"):
+            avatar.icon = ""
+    print("获取角色数据完成")
 
 
 async def dump_avatars(path: Path):
@@ -124,10 +91,13 @@ async def dump_avatars(path: Path):
 
 
 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 = Avatar(**avatar)
+        m = YattaAvatar(**avatar)
         all_avatars.append(m)
         all_avatars_map[m.id] = m
         all_avatars_name[m.name] = m

func/fetch_materials.py

@@ -1,5 +1,5 @@
-import re
 import asyncio
+import re
 from pathlib import Path
 from typing import List, Dict
 
@@ -9,9 +9,9 @@ from bs4 import BeautifulSoup
 
 from func.client import client
 from func.url import info_url
-from models.wiki import Children
 from models.enums import Quality, MaterialType
 from models.material import Material
+from models.wiki import Children
 
 star_map = {
     1: Quality.One,

func/fetch_src.py

@@ -1,7 +1,8 @@
 import shutil
+from pathlib import Path
+
 import aiofiles
 import ujson as jsonlib
-from pathlib import Path
 
 src_dir = Path("src")
 data_dir = Path("data")

main.py

@@ -1,7 +1,7 @@
 import asyncio
 from pathlib import Path
 from func.fetch_all import get_list
-from func.fetch_avatars import fetch_avatars, fetch_avatars_infos, dump_avatars, read_avatars
+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_monsters import fetch_monsters, fetch_monsters_infos, dump_monsters, read_monsters
@@ -32,7 +32,6 @@ async def wiki(
         await read_materials(data_path / "materials.json")
     if override_avatars:
         await fetch_avatars(main_data[0])
-        await fetch_avatars_infos()
         await dump_avatars(data_path / "avatars.json")
     else:
         await read_avatars(data_path / "avatars.json")

models/avatar.py

@@ -1,60 +1,68 @@
-from typing import List
+from typing import List, Optional
+
 from pydantic import BaseModel
-from .enums import Quality, Destiny, Element
-from .material import Material
+
+from .enums import Destiny, Element
 
 
-class AvatarInfo(BaseModel):
-    occupation: str = ""
-    """所属"""
-    faction: str = ""
-    """派系"""
+class YattaAvatarPath(BaseModel):
+    id: str
+    name: str
 
 
-class AvatarItem(BaseModel):
-    item: Material
-    """物品"""
-    count: int
-    """数量"""
+class YattaAvatarTypes(BaseModel):
+    pathType: YattaAvatarPath
+    combatType: YattaAvatarPath
 
 
-class AvatarPromote(BaseModel):
-    required_level: int
-    """突破所需等级"""
-    promote_level: int = 0
-    """突破等级"""
-    max_level: int
-    """解锁的等级上限"""
-
-    coin: int = 0
-    """信用点"""
-    items: List[AvatarItem]
-    """突破所需材料"""
+class YattaAvatarCV(BaseModel):
+    CV_CN: str
+    CV_JP: str
+    CV_KR: str
+    CV_EN: str
 
 
-class AvatarSoul(BaseModel):
+class YattaAvatarFetter(BaseModel):
+    faction: Optional[str]
+    description: Optional[str]
+    cv: Optional[YattaAvatarCV]
+
+
+class YattaAvatarEidolon(BaseModel):
+    id: int
+    rank: int
+    name: Optional[str]
+    description: Optional[str]
+    icon: str
+
+    @property
+    def icon_url(self) -> str:
+        return f"https://api.yatta.top/hsr/assets/UI/skill/{self.icon}.png"
+
+
+class YattaAvatar(BaseModel):
+    id: int
+    """ 角色ID """
     name: str
     """ 名称 """
-    desc: str
-    """ 介绍 """
-
-
-class Avatar(BaseModel):
-    id: int
-    """角色ID"""
-    name: str
-    """名称"""
+    rank: int
+    """ 星级 """
+    types: YattaAvatarTypes
+    """ 角色类型 """
     icon: str
-    """图标"""
-    quality: Quality
-    """品质"""
-    destiny: Destiny
-    """命途"""
-    element: Element
-    """属性"""
-    information: AvatarInfo
-    """角色信息"""
-    promote: List[AvatarPromote]
-    """角色突破数据"""
-    soul: List[AvatarSoul]
-    """角色星魂数据"""
+    """ 图标 """
+    release: int
+    """ 上线时间 """
+    route: str
+    fetter: YattaAvatarFetter
+    eidolons: List[YattaAvatarEidolon]
+
+    @property
+    def destiny(self) -> Destiny:
+        """ 命途 """
+        return Destiny(self.types.pathType.name)
+
+    @property
+    def element(self) -> Element:
+        """ 属性 """
+        return Element(self.types.combatType.name)

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, all_avatars_map
+from func.fetch_avatars import read_avatars, all_avatars_name, dump_avatars, all_avatars_map
 from models.avatar_config import AvatarConfig, AvatarIcon
 from .client import client
 from .url import avatar_config, text_map, base_station_url, avatar_url
@@ -101,15 +101,8 @@ async def fetch_station(configs_map: Dict[str, AvatarConfig]) -> List[AvatarIcon
 async def fix_avatar_config_ktz():
     data_map = {"开拓者·毁灭": (8001, 8002), "开拓者·存护": (8003, 8004)}
     for key, value in data_map.items():
-        one = all_avatars_name[key]
-        one.name = key
-        two = one.copy()
-        one.id = value[0]
-        two.id = value[1]
-        all_avatars.append(two)
-        all_avatars_map[value[0]] = one
-        all_avatars_map[value[1]] = two
-        all_avatars_name[one.name] = one
+        for i in value:
+            all_avatars_map[i].name = key
 
 
 async def fix_avatar_config(text_map_data: Dict[str, str]):
@@ -118,15 +111,6 @@ async def fix_avatar_config(text_map_data: Dict[str, str]):
     print(f"读取到原始数据：{list(configs_map.keys())}")
     data_path = Path("data")
     await read_avatars(data_path / "avatars.json")
-    for key, value in all_avatars_name.items():
-        if key.startswith("开拓者"):
-            continue
-        else:
-            config = configs_map.get(key)
-        if config is None:
-            print(f"错误：未找到角色 {key} 的配置")
-            continue
-        value.id = config.AvatarID
     await fix_avatar_config_ktz()
     icons = await fetch_station(configs_map)
     await dump_icons(data_path / "avatar_icons.json", icons)

res_func/light_cone.py

@@ -7,9 +7,9 @@ import ujson
 from bs4 import BeautifulSoup, Tag
 
 from func.fetch_light_cones import read_light_cones, all_light_cones_name, dump_light_cones
+from models.light_cone_config import LightConeIcon
 from .client import client
 from .url import base_station_url, light_cone_url
-from models.light_cone_config import LightConeIcon
 
 
 async def parse_station(icon: LightConeIcon, tag: Tag):

res_func/relic.py

@@ -4,11 +4,11 @@ from typing import List, Dict
 import aiofiles
 import ujson
 
-from models.enums import RelicAffix, RelicPosition
 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 models.relic_affix import RelicAffixAll, SingleRelicAffix
 
 final_datas: List[RelicAffixAll] = []
 final_datas_map: Dict[str, RelicAffixAll] = {}

res_func/relic_res.py

@@ -1,7 +1,8 @@
-from bs4 import BeautifulSoup
 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 res_func.client import client
 from res_func.url import relic_url, base_station_url

res_func/yatta/avatar.py

@@ -6,9 +6,10 @@ from typing import List
 import aiofiles
 import ujson
 
+from func.fetch_avatars import all_avatars
+from models.avatar import YattaAvatar
 from res_func.client import client
-from res_func.url import avatar_yatta_url, avatar_skill_url
-from res_func.yatta.model import YattaAvatar
+from res_func.url import avatar_skill_url
 
 avatar_data = {}
 avatars_skills_icons = {}
@@ -16,17 +17,11 @@ avatars_skills_path = Path("data/skill")
 avatars_skills_path.mkdir(exist_ok=True, parents=True)
 
 
-async def get_all_avatar() -> List[str]:
-    req = await client.get(avatar_yatta_url)
-    return list(req.json()["data"]["items"].keys())
-
-
 def retry(func):
     async def wrapper(*args, **kwargs):
         for i in range(3):
             try:
-                await func(*args, **kwargs)
-                break
+                return await func(*args, **kwargs)
             except Exception:
                 print(f"重试 {func.__name__} {i + 1} 次")
                 await asyncio.sleep(1)
@@ -34,21 +29,6 @@ def retry(func):
     return wrapper
 
 
-@retry
-async def get_single_avatar(url: str) -> None:
-    req = await client.get(url)
-    try:
-        avatar = YattaAvatar(**req.json()["data"])
-    except Exception as e:
-        print(f"{url} 获取星魂数据失败")
-        raise e
-    if len(avatar.eidolons) != 6:
-        print(f"{url} 获取星魂图片失败")
-        return
-    urls = [i.icon_url for i in avatar.eidolons]
-    avatar_data[str(avatar.id)] = urls
-
-
 @retry
 async def get_single_avatar_skill_icon(url: str, real_path: str) -> None:
     req = await client.get(url)
@@ -77,27 +57,28 @@ async def dump_icons():
 
 async def get_all_avatars() -> None:
     print("开始获取星魂图片")
-    avatar_ids = await get_all_avatar()
-    for avatar_id in avatar_ids:
-        await get_single_avatar(f"{avatar_yatta_url}/{avatar_id}")
+    for avatar in all_avatars:
+        urls = [i.icon_url for i in avatar.eidolons]
+        avatar_data[str(avatar.id)] = urls
     await dump_icons()
     print("获取星魂图片成功")
-    await get_all_avatars_skills_icons(avatar_ids)
+    print("开始获取技能图片")
+    await get_all_avatars_skills_icons(all_avatars)
+    print("获取技能图片成功")
 
 
-async def get_all_avatars_skills_icons(avatar_ids: List[str]):
+async def get_all_avatars_skills_icons(avatars: List[YattaAvatar]):
     remote_path = ["Normal", "BP", "Passive", "Maze", "Ultra"]
     local_path = ["basic_atk", "skill", "talent", "technique", "ultimate"]
-    print("开始获取技能图片")
     tasks = []
-    for avatar_id in avatar_ids:
-        if avatar_id in ["8002", "8004"]:
+    for avatar in avatars:
+        if avatar.id in ["8002", "8004"]:
             continue
         for i in range(len(remote_path)):
             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_skill_url}SkillIcon_{avatar.id}_{remote_path[i]}.png",
+                    f"{avatar.id}_{local_path[i]}.png"
                 )
             )
         await asyncio.gather(*tasks)
@@ -105,4 +86,3 @@ async def get_all_avatars_skills_icons(avatar_ids: List[str]):
     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:
         await f.write(json.dumps(datas, indent=4, ensure_ascii=False))
-    print("获取技能图片成功")

res_func/yatta/model.py

@@ -1,63 +0,0 @@
-from typing import List, Optional
-
-from pydantic import BaseModel, root_validator
-
-from res_func.url import avatar_skill_url
-
-
-class YattaAvatarPath(BaseModel):
-    id: str
-    name: str
-
-
-class YattaAvatarTypes(BaseModel):
-    pathType: YattaAvatarPath
-    combatType: YattaAvatarPath
-
-
-class YattaAvatarCV(BaseModel):
-    CV_CN: str
-    CV_JP: str
-    CV_KR: str
-    CV_EN: str
-
-
-class YattaAvatarFetter(BaseModel):
-    faction: Optional[str]
-    description: Optional[str]
-    cv: Optional[YattaAvatarCV]
-
-
-class YattaAvatarEidolon(BaseModel):
-    id: int
-    rank: int
-    name: str
-    description: str
-    icon: str
-
-    @property
-    def icon_url(self) -> str:
-        return f"{avatar_skill_url}{self.icon}.png"
-
-
-class YattaAvatar(BaseModel):
-    id: int
-    name: str
-    rank: int
-    types: YattaAvatarTypes
-    icon: str
-    release: int
-    route: str
-    fetter: YattaAvatarFetter
-    eidolons: List[YattaAvatarEidolon]
-
-    @root_validator(pre=True)
-    def validate(cls, values):
-        if values.get("eidolons") is None:
-            values["eidolons"] = []
-        else:
-            eidolons = []
-            for eidolon in values["eidolons"].values():
-                eidolons.append(eidolon)
-            values["eidolons"] = eidolons
-        return values