PaiGramTeam/PamGram
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/PamGram.git synced 2024-11-21 21:58:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

♻️ Refactor wiki models

Browse Source
This commit is contained in:
xtaodada 2023-08-29 14:03:21 +08:00
parent e330904d77
commit 735d420ae5
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
17 changed files with 201 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
core/services/game/cache.py
View File

@ -3,7 +3,14 @@ from typing import List
from core.base_service import BaseService
from core.dependence.redisdb import RedisDB
__all__ = ["GameCache", "GameCacheForStrategy", "GameCacheForMaterial", "GameCacheForLightCone", "GameCacheForRelics"]
__all__ = [
"GameCache",
"GameCacheForAvatar",
"GameCacheForStrategy",
"GameCacheForMaterial",
"GameCacheForLightCone",
"GameCacheForRelics",
]
class GameCache:
@ -34,6 +41,10 @@ class GameCache:
await self.client.expire(qname, self.ttl)
class GameCacheForAvatar(BaseService.Component, GameCache):
qname = "game:avatar"
class GameCacheForStrategy(BaseService.Component, GameCache):
qname = "game:strategy"

11
core/services/game/services.py
View File

@ -1,5 +1,6 @@
from core.base_service import BaseService
from core.services.game.cache import (
GameCacheForAvatar,
GameCacheForStrategy,
GameCacheForMaterial,
GameCacheForLightCone,
@ -12,16 +13,26 @@ __all__ = "GameCacheService"
class GameCacheService(BaseService):
def __init__(
self,
avatar_cache: GameCacheForAvatar,
strategy_cache: GameCacheForStrategy,
material_cache: GameCacheForMaterial,
light_cone_cache: GameCacheForLightCone,
relics_cache: GameCacheForRelics,
):
self.avatar_cache = avatar_cache
self.strategy_cache = strategy_cache
self.material_cache = material_cache
self.light_cone_cache = light_cone_cache
self.relics_cache = relics_cache
async def get_avatar_cache(self, character_name: str) -> str:
cache = await self.avatar_cache.get_file(character_name)
if cache is not None:
return cache.decode("utf-8")
async def set_avatar_cache(self, character_name: str, file: str) -> None:
await self.avatar_cache.set_file(character_name, file)
async def get_strategy_cache(self, character_name: str) -> str:
cache = await self.strategy_cache.get_file(character_name)
if cache is not None:

9
core/services/wiki/services.py
View File

@ -1,9 +1,6 @@
from typing import NoReturn
from core.base_service import BaseService
from modules.wiki.character import Character
from modules.wiki.material import Material
from modules.wiki.monster import Monster
from modules.wiki.relic import Relic
from modules.wiki.light_cone import LightCone
from modules.wiki.raider import Raider
@ -16,7 +13,6 @@ class WikiService(BaseService):
def __init__(self):
self.character = Character()
self.material = Material()
self.monster = Monster()
self.relic = Relic()
self.light_cone = LightCone()
self.raider = Raider()
@ -26,7 +22,6 @@ class WikiService(BaseService):
try:
await self.character.read()
await self.material.read()
await self.monster.read()
await self.relic.read()
await self.light_cone.read()
await self.raider.read()
@ -34,14 +29,12 @@ class WikiService(BaseService):
logger.error("加载 Wiki 数据失败", exc_info=e)
logger.info("加载 Wiki 数据完成")
async def refresh_wiki(self) -> NoReturn:
async def refresh_wiki(self) -> None:
logger.info("正在重新获取Wiki")
logger.info("正在重新获取角色信息")
await self.character.refresh()
logger.info("正在重新获取材料信息")
await self.material.refresh()
logger.info("正在重新获取敌对生物信息")
await self.monster.refresh()
logger.info("正在重新获取遗器信息")
await self.relic.refresh()
logger.info("正在重新获取光锥信息")

14
modules/wiki/light_cone.py
View File

@ -1,7 +1,7 @@
from typing import List, Dict, Optional
from modules.wiki.base import WikiModel
from modules.wiki.models.light_cone import LightCone as LightConeModel
from modules.wiki.models.light_cone import YattaLightCone
class LightCone(WikiModel):
@ -10,9 +10,9 @@ class LightCone(WikiModel):
def __init__(self):
super().__init__()
self.all_light_cones: List[LightConeModel] = []
self.all_light_cones_map: Dict[int, LightConeModel] = {}
self.all_light_cones_name: Dict[str, LightConeModel] = {}
self.all_light_cones: List[YattaLightCone] = []
self.all_light_cones_map: Dict[int, YattaLightCone] = {}
self.all_light_cones_name: Dict[str, YattaLightCone] = {}
def clear_class_data(self) -> None:
self.all_light_cones.clear()
@ -31,15 +31,15 @@ class LightCone(WikiModel):
datas = await WikiModel.read(self.light_cone_path)
self.clear_class_data()
for data in datas:
m = LightConeModel(**data)
m = YattaLightCone(**data)
self.all_light_cones.append(m)
self.all_light_cones_map[m.id] = m
self.all_light_cones_name[m.name] = m
def get_by_id(self, cid: int) -> Optional[LightConeModel]:
def get_by_id(self, cid: int) -> Optional[YattaLightCone]:
return self.all_light_cones_map.get(cid)
def get_by_name(self, name: str) -> Optional[LightConeModel]:
def get_by_name(self, name: str) -> Optional[YattaLightCone]:
return self.all_light_cones_name.get(name)
def get_name_list(self) -> List[str]:

14
modules/wiki/material.py
View File

@ -1,7 +1,7 @@
from typing import List, Dict, Optional
from modules.wiki.base import WikiModel
from modules.wiki.models.material import Material as MaterialModel
from modules.wiki.models.material import YattaMaterial
class Material(WikiModel):
@ -10,9 +10,9 @@ class Material(WikiModel):
def __init__(self):
super().__init__()
self.all_materials: List[MaterialModel] = []
self.all_materials_map: Dict[int, MaterialModel] = {}
self.all_materials_name: Dict[str, MaterialModel] = {}
self.all_materials: List[YattaMaterial] = []
self.all_materials_map: Dict[int, YattaMaterial] = {}
self.all_materials_name: Dict[str, YattaMaterial] = {}
def clear_class_data(self) -> None:
self.all_materials.clear()
@ -31,15 +31,15 @@ class Material(WikiModel):
datas = await WikiModel.read(self.material_path)
self.clear_class_data()
for data in datas:
m = MaterialModel(**data)
m = YattaMaterial(**data)
self.all_materials.append(m)
self.all_materials_map[m.id] = m
self.all_materials_name[m.name] = m
def get_by_id(self, cid: int) -> Optional[MaterialModel]:
def get_by_id(self, cid: int) -> Optional[YattaMaterial]:
return self.all_materials_map.get(cid)
def get_by_name(self, name: str) -> Optional[MaterialModel]:
def get_by_name(self, name: str) -> Optional[YattaMaterial]:
return self.all_materials_name.get(name)
def get_name_list(self) -> List[str]:

48
modules/wiki/models/light_cone.py
View File

@ -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

32
modules/wiki/models/material.py
View File

@ -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"

25
modules/wiki/models/monster.py
View File

@ -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
"""发现地点"""

9
modules/wiki/models/relic.py
View File

@ -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

46
modules/wiki/monster.py
View File

@ -1,46 +0,0 @@
from typing import List, Dict, Optional
from modules.wiki.base import WikiModel
from modules.wiki.models.monster import Monster as MonsterModel
class Monster(WikiModel):
monster_url = WikiModel.BASE_URL + "monsters.json"
monster_path = WikiModel.BASE_PATH / "monsters.json"
def __init__(self):
super().__init__()
self.all_monsters: List[MonsterModel] = []
self.all_monsters_map: Dict[int, MonsterModel] = {}
self.all_monsters_name: Dict[str, MonsterModel] = {}
def clear_class_data(self) -> None:
self.all_monsters.clear()
self.all_monsters_map.clear()
self.all_monsters_name.clear()
async def refresh(self):
datas = await self.remote_get(self.monster_url)
await self.dump(datas.json(), self.monster_path)
await self.read()
async def read(self):
if not self.monster_path.exists():
await self.refresh()
return
datas = await WikiModel.read(self.monster_path)
self.clear_class_data()
for data in datas:
m = MonsterModel(**data)
self.all_monsters.append(m)
self.all_monsters_map[m.id] = m
self.all_monsters_name[m.name] = m
def get_by_id(self, cid: int) -> Optional[MonsterModel]:
return self.all_monsters_map.get(cid)
def get_by_name(self, name: str) -> Optional[MonsterModel]:
return self.all_monsters_name.get(name)
def get_name_list(self) -> List[str]:
return list(self.all_monsters_name.keys())

13
modules/wiki/raider.py
View File

@ -7,25 +7,35 @@ class Raider(WikiModel):
raider_url = "https://raw.githubusercontent.com/PaiGramTeam/star-rail-atlas/master"
raider_path = WikiModel.BASE_PATH / "raiders"
raider_role_path = WikiModel.BASE_PATH / "raiders" / "role"
raider_guide_for_role_path = WikiModel.BASE_PATH / "raiders" / "guide_for_role"
raider_light_cone_path = WikiModel.BASE_PATH / "raiders" / "light_cone"
raider_role_material_path = WikiModel.BASE_PATH / "raiders" / "role_material"
raider_relic_path = WikiModel.BASE_PATH / "raiders" / "relic"
raider_info_path = WikiModel.BASE_PATH / "raiders" / "path.json"
raider_role_path.mkdir(parents=True, exist_ok=True)
raider_guide_for_role_path.mkdir(parents=True, exist_ok=True)
raider_light_cone_path.mkdir(parents=True, exist_ok=True)
raider_role_material_path.mkdir(parents=True, exist_ok=True)
raider_relic_path.mkdir(parents=True, exist_ok=True)
name_map = {"role": "role", "lightcone": "light_cone", "material for role": "role_material", "relic": "relic"}
name_map = {
"role": "role",
"lightcone": "light_cone",
"material for role": "role_material",
"relic": "relic",
"guide for role": "guide_for_role",
}
def __init__(self):
super().__init__()
self.all_role_raiders: List[str] = []
self.all_guide_for_role_raiders: List[str] = []
self.all_light_cone_raiders: List[str] = []
self.all_role_material_raiders: List[str] = []
self.all_relic_raiders: List[str] = []
def clear_class_data(self) -> None:
self.all_role_raiders.clear()
self.all_guide_for_role_raiders.clear()
self.all_light_cone_raiders.clear()
self.all_role_material_raiders.clear()
self.all_relic_raiders.clear()
@ -54,6 +64,7 @@ class Raider(WikiModel):
datas: Dict[str, List] = await WikiModel.read(self.raider_info_path) # noqa
self.clear_class_data()
self.all_role_raiders.extend(datas["role"])
self.all_guide_for_role_raiders.extend(datas["guide_for_role"])
self.all_light_cone_raiders.extend(datas["light_cone"])
self.all_role_material_raiders.extend(datas["role_material"])
self.all_relic_raiders.extend(datas["relic"])

14
modules/wiki/relic.py
View File

@ -1,7 +1,7 @@
from typing import List, Dict, Optional
from modules.wiki.base import WikiModel
from modules.wiki.models.relic import Relic as RelicModel
from modules.wiki.models.relic import YattaRelic
class Relic(WikiModel):
@ -10,9 +10,9 @@ class Relic(WikiModel):
def __init__(self):
super().__init__()
self.all_relics: List[RelicModel] = []
self.all_relics_map: Dict[int, RelicModel] = {}
self.all_relics_name: Dict[str, RelicModel] = {}
self.all_relics: List[YattaRelic] = []
self.all_relics_map: Dict[int, YattaRelic] = {}
self.all_relics_name: Dict[str, YattaRelic] = {}
def clear_class_data(self) -> None:
self.all_relics.clear()
@ -31,15 +31,15 @@ class Relic(WikiModel):
datas = await WikiModel.read(self.relic_path)
self.clear_class_data()
for data in datas:
m = RelicModel(**data)
m = YattaRelic(**data)
self.all_relics.append(m)
self.all_relics_map[m.id] = m
self.all_relics_name[m.name] = m
def get_by_id(self, cid: int) -> Optional[RelicModel]:
def get_by_id(self, cid: int) -> Optional[YattaRelic]:
return self.all_relics_map.get(cid)
def get_by_name(self, name: str) -> Optional[RelicModel]:
def get_by_name(self, name: str) -> Optional[YattaRelic]:
return self.all_relics_name.get(name)
def get_name_list(self) -> List[str]:

4
plugins/admin/wiki.py
View File

@ -14,6 +14,6 @@ class WikiPlugin(Plugin):
@handler.command("refresh_wiki", block=False, admin=True)
async def refresh_wiki(self, update: Update, _: CallbackContext):
message = update.effective_message
await message.reply_text("正在刷新Wiki缓存，请稍等")
msg = await message.reply_text("正在刷新Wiki缓存，请稍等")
await self.wiki_service.refresh_wiki()
await message.reply_text("刷新Wiki缓存成功")
await msg.edit_text("刷新Wiki缓存成功")

17
plugins/app/inline.py
View File

@ -35,6 +35,7 @@ class Inline(Plugin):
self.weapons_list: List[Dict[str, str]] = []
self.characters_list: List[Dict[str, str]] = []
self.characters_material_list: List[Dict[str, str]] = []
self.characters_guide_list: List[Dict[str, str]] = []
self.light_cone_list: List[Dict[str, str]] = []
self.relics_list: List[Dict[str, str]] = []
self.refresh_task: List[Awaitable] = []
@ -89,6 +90,15 @@ class Inline(Plugin):
if character.startswith(key) or character.endswith(key):
self.characters_material_list.append({"name": character, "icon": value})
break
# 角色攻略
for character in self.wiki_service.raider.all_guide_for_role_raiders:
if character in datas:
self.characters_guide_list.append({"name": character, "icon": datas[character]})
else:
for key, value in datas.items():
if character.startswith(key) or character.endswith(key):
self.characters_guide_list.append({"name": character, "icon": value})
break
logger.success("Inline 模块获取角色列表成功")
self.refresh_task.append(asyncio.create_task(task_characters()))
@ -108,6 +118,8 @@ class Inline(Plugin):
temp_data = [
("光锥图鉴查询", "输入光锥名称即可查询光锥图鉴"),
("角色攻略查询", "输入角色名即可查询角色攻略图鉴"),
("角色图鉴查询", "输入角色名即可查询角色图鉴"),
("角色培养素材查询", "输入角色名即可查询角色培养素材图鉴"),
("遗器套装查询", "输入遗器套装名称即可查询遗器套装图鉴"),
]
for i in temp_data:
@ -120,11 +132,12 @@ class Inline(Plugin):
)
)
else:
if args[0] in ["查看角色攻略列表并查询", "查看角色培养素材列表并查询", "查看光锥列表并查询", "查看遗器套装列表并查询"]:
if args[0] in ["查看角色攻略列表并查询", "查看角色图鉴列表并查询", "查看光锥列表并查询", "查看遗器套装列表并查询", "查看角色培养素材列表并查询"]:
temp_data = {
"查看角色攻略列表并查询": (self.characters_list, "角色攻略查询"),
"查看角色图鉴列表并查询": (self.characters_guide_list, "角色图鉴查询"),
"查看角色培养素材列表并查询": (self.characters_material_list, "角色培养素材查询"),
"查看光锥列表并查询": (self.light_cone_list, "光锥查询"),
"查看光锥列表并查询": (self.light_cone_list, "光锥图鉴查询"),
"查看遗器套装列表并查询": (self.relics_list, "遗器套装查询"),
}[args[0]]
for character in temp_data[0]:

85
plugins/starrail/avatars.py Normal file
View File

@ -0,0 +1,85 @@
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ChatAction, ParseMode
from telegram.ext import CallbackContext, filters
from core.plugin import Plugin, handler
from core.services.game.services import GameCacheService
from core.services.search.models import StrategyEntry
from core.services.search.services import SearchServices
from core.services.wiki.services import WikiService
from metadata.shortname import roleToName, roleToTag
from utils.log import logger
class AvatarsPlugin(Plugin):
"""角色图鉴查询"""
KEYBOARD = [[InlineKeyboardButton(text="查看角色图鉴列表并查询", switch_inline_query_current_chat="查看角色图鉴列表并查询")]]
def __init__(
self,
cache_service: GameCacheService = None,
wiki_service: WikiService = None,
search_service: SearchServices = None,
):
self.cache_service = cache_service
self.wiki_service = wiki_service
self.search_service = search_service
@handler.command(command="avatars", block=False)
@handler.message(filters=filters.Regex("^角色图鉴查询(.*)"), block=False)
async def command_start(self, update: Update, context: CallbackContext) -> None:
message = update.effective_message
user = update.effective_user
args = self.get_args(context)
if len(args) >= 1:
character_name = args[0]
else:
reply_message = await message.reply_text("请回复你要查询的图鉴的角色名", reply_markup=InlineKeyboardMarkup(self.KEYBOARD))
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message)
self.add_delete_message_job(reply_message)
return
character_name = roleToName(character_name)
file_path = self.wiki_service.raider.raider_role_path / f"{character_name}.png"
if not file_path.exists():
reply_message = await message.reply_text(
f"没有找到 {character_name} 的图鉴", reply_markup=InlineKeyboardMarkup(self.KEYBOARD)
)
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message)
self.add_delete_message_job(reply_message)
return
logger.info("用户 %s[%s] 查询角色图鉴命令请求 || 参数 %s", user.full_name, user.id, character_name)
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
caption = "From 米游社@听语惊花"
if file_id := await self.cache_service.get_avatar_cache(character_name):
await message.reply_photo(
photo=file_id,
caption=caption,
filename=f"{character_name}.png",
allow_sending_without_reply=True,
parse_mode=ParseMode.HTML,
)
else:
reply_photo = await message.reply_photo(
photo=open(file_path, "rb"),
caption=caption,
filename=f"{character_name}.png",
allow_sending_without_reply=True,
parse_mode=ParseMode.HTML,
)
if reply_photo.photo:
tags = roleToTag(character_name)
photo_file_id = reply_photo.photo[0].file_id
await self.cache_service.set_avatar_cache(character_name, photo_file_id)
entry = StrategyEntry(
key=f"plugin:avatar:{character_name}",
title=character_name,
description=f"{character_name} 角色图鉴",
tags=tags,
caption=caption,
parse_mode="HTML",
photo_file_id=photo_file_id,
)
await self.search_service.add_entry(entry)

2
plugins/starrail/light_cone.py
View File

@ -27,7 +27,7 @@ class LightConePlugin(Plugin):
self.search_service = search_service
@handler.command(command="light_cone", block=False)
@handler.message(filters=filters.Regex("^光锥查询(.*)"), block=False)
@handler.message(filters=filters.Regex("^光锥图鉴查询(.*)"), block=False)
async def command_start(self, update: Update, context: CallbackContext) -> None:
message = update.effective_message
user = update.effective_user

2
plugins/starrail/strategy.py
View File

@ -41,7 +41,7 @@ class StrategyPlugin(Plugin):
self.add_delete_message_job(reply_message)
return
character_name = roleToName(character_name)
file_path = self.wiki_service.raider.raider_role_path / f"{character_name}.png"
file_path = self.wiki_service.raider.raider_guide_for_role_path / f"{character_name}.png"
if not file_path.exists():
reply_message = await message.reply_text(
f"没有找到 {character_name} 的攻略", reply_markup=InlineKeyboardMarkup(self.KEYBOARD)