Support role combat plugin

This commit is contained in:
omg-xtao 2024-07-29 19:06:00 +08:00 committed by GitHub
parent 6913a81206
commit 3509f9e441
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 1216 additions and 10 deletions

View File

@ -3,6 +3,7 @@ from typing import Dict
from pydantic import BaseModel
from simnet.models.genshin.chronicle.abyss import SpiralAbyss
from simnet.models.genshin.chronicle.img_theater import ImgTheaterData
from simnet.models.genshin.diary import Diary
from gram_core.services.history_data.models import HistoryData
@ -12,12 +13,14 @@ __all__ = (
"HistoryDataTypeEnum",
"HistoryDataAbyss",
"HistoryDataLedger",
"HistoryDataImgTheater",
)
class HistoryDataTypeEnum(int, enum.Enum):
ABYSS = 0 # 深境螺旋
LEDGER = 2 # 开拓月历
ROLE_COMBAT = 3 # 幻想真境剧诗
class HistoryDataAbyss(BaseModel):
@ -35,3 +38,12 @@ class HistoryDataLedger(BaseModel):
@classmethod
def from_data(cls, data: HistoryData) -> "HistoryDataLedger":
return cls.parse_obj(data.data)
class HistoryDataImgTheater(BaseModel):
abyss_data: ImgTheaterData
character_data: Dict[int, int]
@classmethod
def from_data(cls, data: HistoryData) -> "HistoryDataImgTheater":
return cls.parse_obj(data.data)

View File

@ -3,9 +3,16 @@ from typing import Dict, List
from pytz import timezone
from simnet.models.genshin.chronicle.abyss import SpiralAbyss
from simnet.models.genshin.chronicle.img_theater import ImgTheaterData
from simnet.models.genshin.diary import Diary
from core.services.history_data.models import HistoryData, HistoryDataTypeEnum, HistoryDataAbyss, HistoryDataLedger
from core.services.history_data.models import (
HistoryData,
HistoryDataTypeEnum,
HistoryDataAbyss,
HistoryDataLedger,
HistoryDataImgTheater,
)
from gram_core.base_service import BaseService
from gram_core.services.history_data.services import HistoryDataBaseServices
@ -19,6 +26,7 @@ __all__ = (
"HistoryDataBaseServices",
"HistoryDataAbyssServices",
"HistoryDataLedgerServices",
"HistoryDataImgTheaterServices",
)
TZ = timezone("Asia/Shanghai")
@ -65,3 +73,24 @@ class HistoryDataLedgerServices(BaseService, HistoryDataBaseServices):
type=HistoryDataLedgerServices.DATA_TYPE,
data=jsonlib.loads(json_data),
)
class HistoryDataImgTheaterServices(BaseService, HistoryDataBaseServices):
DATA_TYPE = HistoryDataTypeEnum.ROLE_COMBAT.value
@staticmethod
def exists_data(data: HistoryData, old_data: List[HistoryData]) -> bool:
floor = data.data.get("detail", {}).get("rounds_data")
return any(d.data.get("detail", {}).get("rounds_data") == floor for d in old_data)
@staticmethod
def create(user_id: int, abyss_data: ImgTheaterData, character_data: Dict[int, int]):
data = HistoryDataImgTheater(abyss_data=abyss_data, character_data=character_data)
json_data = data.json(by_alias=True, encoder=json_encoder)
return HistoryData(
user_id=user_id,
data_id=abyss_data.schedule.id,
time_created=datetime.datetime.now(),
type=HistoryDataImgTheaterServices.DATA_TYPE,
data=jsonlib.loads(json_data),
)

View File

@ -2168,7 +2168,7 @@ name = "simnet"
version = "0.1.22"
requires_python = "<4.0,>=3.8"
git = "https://github.com/PaiGramTeam/SIMNet"
revision = "d27e6e9e46e48e32ce70abb19a4030a1a93c940c"
revision = "0ddc9dabd1543450a7c3f513f177882ed04425fd"
summary = "Modern API wrapper for Genshin Impact & Honkai: Star Rail built on asyncio and pydantic."
groups = ["default"]
dependencies = [

View File

@ -66,6 +66,8 @@ class SetCommandPlugin(Plugin):
BotCommand("abyss", "查询深渊战绩"),
BotCommand("abyss_team", "查询深渊推荐配队"),
BotCommand("abyss_history", "查询深渊历史战绩"),
BotCommand("role_combat", "查询幻想真境剧诗战绩"),
BotCommand("role_combat_history", "查询幻想真境剧诗历史战绩"),
BotCommand("avatars", "查询角色练度"),
BotCommand("reg_time", "账号注册时间"),
BotCommand("daily_material", "今日素材表"),

View File

@ -0,0 +1,587 @@
"""幻想真境剧诗数据查询"""
import asyncio
import math
import re
from functools import lru_cache, partial
from typing import Any, Coroutine, List, Optional, Tuple, Union, Dict
from arkowrapper import ArkoWrapper
from pytz import timezone
from simnet import GenshinClient
from simnet.models.genshin.chronicle.img_theater import ImgTheaterData, TheaterDifficulty
from telegram import Message, Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ChatAction, ParseMode
from telegram.ext import CallbackContext, filters, ContextTypes
from core.dependence.assets import AssetsService
from core.plugin import Plugin, handler
from core.services.cookies.error import TooManyRequestPublicCookies
from core.services.history_data.models import HistoryDataImgTheater
from core.services.history_data.services import HistoryDataImgTheaterServices
from core.services.template.models import RenderGroupResult, RenderResult
from core.services.template.services import TemplateService
from gram_core.config import config
from gram_core.dependence.redisdb import RedisDB
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from plugins.tools.genshin import GenshinHelper
from utils.enkanetwork import RedisCache
from utils.log import logger
from utils.uid import mask_number
TZ = timezone("Asia/Shanghai")
get_args_pattern = re.compile(r"\d+")
@lru_cache
def get_args(text: str) -> Tuple[int, bool, bool]:
total = "all" in text or "总览" in text
prev = "pre" in text or "上期" in text
floor = 0
if not total:
m = get_args_pattern.search(text)
if m is not None:
floor = int(m.group(0))
return floor, total, prev
class AbyssUnlocked(Exception):
"""根本没动"""
class NoMostKills(Exception):
"""挑战了但是数据没刷新"""
class FloorNotFoundError(Exception):
"""只有数据统计,幕数统计未出"""
class AbyssNotFoundError(Exception):
"""如果查询别人,是无法找到队伍详细,只有数据统计"""
class RoleCombatPlugin(Plugin):
"""幻想真境剧诗数据查询"""
def __init__(
self,
template: TemplateService,
helper: GenshinHelper,
assets_service: AssetsService,
history_data_abyss: HistoryDataImgTheaterServices,
redis: RedisDB,
):
self.template_service = template
self.helper = helper
self.assets_service = assets_service
self.history_data_abyss = history_data_abyss
self.cache = RedisCache(redis.client, key="plugin:role_combat:history")
@handler.command("role_combat", block=False)
@handler.message(filters.Regex(r"^幻想真境剧诗数据"), block=False)
async def command_start(self, update: Update, context: CallbackContext) -> None: # skipcq: PY-R1000 #
user_id = await self.get_real_user_id(update)
uid, offset = self.get_real_uid_or_offset(update)
args = self.get_args(context)
message = update.effective_message
# 若查询帮助
if (message.text.startswith("/") and "help" in message.text) or "帮助" in message.text:
await message.reply_text(
"<b>幻想真境剧诗挑战数据</b>功能使用帮助(中括号表示可选参数)\n\n"
"指令格式:\n<code>/role_combat + [幕数/all] + [pre]</code>\n<code>pre</code>表示上期)\n\n"
"文本格式:\n<code>幻想真境剧诗数据 + 查询/总览 + [上期] + [幕数]</code> \n\n"
"例如以下指令都正确:\n"
"<code>/role_combat</code>\n<code>/role_combat 6 pre</code>\n<code>/role_combat all pre</code>\n"
"<code>幻想真境剧诗数据查询</code>\n<code>幻想真境剧诗数据查询上期第6幕</code>\n<code>幻想真境剧诗数据总览上期</code>",
parse_mode=ParseMode.HTML,
)
self.log_user(update, logger.info, "查询[bold]幻想真境剧诗挑战数据[/bold]帮助", extra={"markup": True})
return
# 解析参数
floor, total, previous = get_args(" ".join([i for i in args if not i.startswith("@")]))
if floor > 8 or floor < 0:
reply_msg = await message.reply_text("幻想真境剧诗幕数输入错误,请重新输入。支持的参数为: 1-8 或 all")
if filters.ChatType.GROUPS.filter(message):
self.add_delete_message_job(reply_msg)
self.add_delete_message_job(message)
return
self.log_user(
update,
logger.info,
"[bold]幻想真境剧诗挑战数据[/bold]请求: floor=%s total=%s previous=%s",
floor,
total,
previous,
extra={"markup": True},
)
await message.reply_chat_action(ChatAction.TYPING)
reply_text: Optional[Message] = None
if total:
reply_text = await message.reply_text(
f"{config.notice.bot_name}需要时间整理幻想真境剧诗数据,还请耐心等待哦~"
)
try:
async with self.helper.genshin_or_public(user_id, uid=uid, offset=offset) as client:
if not client.public:
await client.get_record_cards()
abyss_data, avatar_data = await self.get_rendered_pic_data(client, client.player_id, previous)
images = await self.get_rendered_pic(abyss_data, avatar_data, client.player_id, floor, total)
except AbyssUnlocked: # 若幻想真境剧诗未解锁
await message.reply_text("还未解锁幻想真境剧诗哦~")
return
except NoMostKills: # 若幻想真境剧诗还未挑战
await message.reply_text("还没有挑战本次幻想真境剧诗呢,咕咕咕~")
return
except FloorNotFoundError:
await message.reply_text("幻想真境剧诗详细数据未找到,咕咕咕~")
return
except AbyssNotFoundError:
await message.reply_text("无法查询玩家挑战队伍详情,只能查询统计详情哦~")
return
except TooManyRequestPublicCookies:
reply_message = await message.reply_text("查询次数太多,请您稍后重试")
if filters.ChatType.GROUPS.filter(message):
self.add_delete_message_job(reply_message)
self.add_delete_message_job(message)
return
finally:
if reply_text is not None:
await reply_text.delete()
if images is None:
await message.reply_text(f"还没有第 {floor} 幕的挑战数据")
return
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
for group in ArkoWrapper(images).group(10): # 每 10 张图片分一个组
await RenderGroupResult(results=group).reply_media_group(message, write_timeout=60)
self.log_user(update, logger.info, "[bold]幻想真境剧诗挑战数据[/bold]: 成功发送图片", extra={"markup": True})
async def get_rendered_pic_data(
self, client: GenshinClient, uid: int, previous: bool
) -> Tuple["ImgTheaterData", Dict[int, int]]:
abyss_data = await client.get_genshin_imaginarium_theater(
uid, need_detail=not client.public, previous=previous, lang="zh-cn"
) # noqa
avatar_data = {}
if (not abyss_data.unlocked) or (not abyss_data.data):
raise AbyssUnlocked
abyss_data = abyss_data.data[0]
if not client.public: # noqa
avatars = await client.get_genshin_characters(uid, lang="zh-cn")
avatar_data = {i.id: i.constellation for i in avatars}
if abyss_data.has_data and abyss_data.has_detail_data and abyss_data.detail:
await self.save_abyss_data(self.history_data_abyss, uid, abyss_data, avatar_data)
return abyss_data, avatar_data
async def get_rendered_pic( # skipcq: PY-R1000 #
self, abyss_data: "ImgTheaterData", avatar_data: Dict[int, int], uid: int, floor: int, total: bool
) -> Union[Tuple[Any], List[RenderResult], None]:
"""
获取渲染后的图片
Args:
abyss_data (ImgTheaterData): 幻想真境剧诗数据
avatar_data (Dict[int, int]): 角色数据
uid (int): 需要查询的 uid
floor (int): 幕数
total (bool): 是否为总览
Returns:
bytes格式的图片
"""
if (total or (floor > 0)) and (not abyss_data.detail or len(abyss_data.detail.rounds_data) == 0):
raise FloorNotFoundError
if (total or (floor > 0)) and not abyss_data.detail:
raise AbyssNotFoundError
start_time = abyss_data.schedule.start_time.astimezone(TZ)
time = start_time.strftime("%Y年%m月")
render_data = {
"time": time,
"stat": abyss_data.stat,
"uid": mask_number(uid),
"floor_colors": {
1: "#374952",
2: "#374952",
3: "#55464B",
4: "#55464B",
5: "#55464B",
6: "#1D2A5D",
7: "#1D2A5D",
8: "#1D2A5D",
9: "#292B58",
10: "#382024",
11: "#252550",
12: "#1D2A4A",
},
}
if total:
render_data["avatar_data"] = avatar_data
render_inputs: List[Tuple[int, Coroutine[Any, Any, RenderResult]]] = []
def overview_task():
return -1, self.template_service.render(
"genshin/role_combat/overview.jinja2", render_data, viewport={"width": 750, "height": 320}
)
def floor_task(floor_index: int):
floor_d = abyss_data.detail.rounds_data[floor_index]
return (
floor_d.round_id,
self.template_service.render(
"genshin/role_combat/floor.jinja2",
{
**render_data,
"floor": floor_d,
"floor_time": floor_d.finish_time.astimezone(TZ).strftime("%Y-%m-%d %H:%M:%S"),
},
viewport={"width": 690, "height": 500},
full_page=True,
ttl=15 * 24 * 60 * 60,
),
)
render_inputs.append(overview_task())
for i, _ in enumerate(abyss_data.detail.rounds_data):
render_inputs.append(floor_task(i))
render_group_inputs = list(map(lambda x: x[1], sorted(render_inputs, key=lambda x: x[0])))
return await asyncio.gather(*render_group_inputs)
if floor < 1:
return [
await self.template_service.render(
"genshin/role_combat/overview.jinja2", render_data, viewport={"width": 750, "height": 320}
)
]
if not (floor_data := list(filter(lambda x: x.round_id == floor, abyss_data.detail.rounds_data))):
return None
render_data["avatar_data"] = avatar_data
render_data["floor"] = floor_data[0]
render_data["floor_time"] = floor_data[0].finish_time.astimezone(TZ).strftime("%Y-%m-%d %H:%M:%S")
return [
await self.template_service.render(
"genshin/role_combat/floor.jinja2", render_data, viewport={"width": 690, "height": 500}
)
]
@staticmethod
async def save_abyss_data(
history_data_abyss: "HistoryDataImgTheaterServices",
uid: int,
abyss_data: "ImgTheaterData",
character_data: Dict[int, int],
) -> bool:
model = history_data_abyss.create(uid, abyss_data, character_data)
old_data = await history_data_abyss.get_by_user_id_data_id(uid, model.data_id)
exists = history_data_abyss.exists_data(model, old_data)
if not exists:
await history_data_abyss.add(model)
return True
return False
async def get_abyss_data(self, uid: int):
return await self.history_data_abyss.get_by_user_id(uid)
@staticmethod
def get_season_data_name(data: "HistoryDataImgTheater"):
start_time = data.abyss_data.schedule.start_time.astimezone(TZ)
time = start_time.strftime("%Y.%m ")[2:]
honor = ""
if data.abyss_data.stat.difficulty == TheaterDifficulty.EASY:
diff = "简单"
elif data.abyss_data.stat.difficulty == TheaterDifficulty.NORMAL:
diff = "普通"
else:
diff = "困难"
if data.abyss_data.stat.medal_num == 8 and data.abyss_data.stat.difficulty == TheaterDifficulty.HARD:
honor = "👑"
return f"{time} {data.abyss_data.stat.medal_num}{diff} {honor}"
async def get_session_button_data(self, user_id: int, uid: int, force: bool = False):
redis = await self.cache.get(str(uid))
if redis and not force:
return redis["buttons"]
data = await self.get_abyss_data(uid)
data.sort(key=lambda x: x.id, reverse=True)
abyss_data = [HistoryDataImgTheater.from_data(i) for i in data]
buttons = [
{
"name": RoleCombatPlugin.get_season_data_name(abyss_data[idx]),
"value": f"get_role_combat_history|{user_id}|{uid}|{value.id}",
}
for idx, value in enumerate(data)
]
await self.cache.set(str(uid), {"buttons": buttons})
return buttons
async def gen_season_button(
self,
user_id: int,
uid: int,
page: int = 1,
) -> List[List[InlineKeyboardButton]]:
"""生成按钮"""
data = await self.get_session_button_data(user_id, uid)
if not data:
return []
buttons = [
InlineKeyboardButton(
value["name"],
callback_data=value["value"],
)
for value in data
]
all_buttons = [buttons[i : i + 3] for i in range(0, len(buttons), 3)]
send_buttons = all_buttons[(page - 1) * 5 : page * 5]
last_page = page - 1 if page > 1 else 0
all_page = math.ceil(len(all_buttons) / 5)
next_page = page + 1 if page < all_page and all_page > 1 else 0
last_button = []
if last_page:
last_button.append(
InlineKeyboardButton(
"<< 上一页",
callback_data=f"get_role_combat_history|{user_id}|{uid}|p_{last_page}",
)
)
if last_page or next_page:
last_button.append(
InlineKeyboardButton(
f"{page}/{all_page}",
callback_data=f"get_role_combat_history|{user_id}|{uid}|empty_data",
)
)
if next_page:
last_button.append(
InlineKeyboardButton(
"下一页 >>",
callback_data=f"get_role_combat_history|{user_id}|{uid}|p_{next_page}",
)
)
if last_button:
send_buttons.append(last_button)
return send_buttons
@staticmethod
async def gen_floor_button(
data_id: int,
abyss_data: "HistoryDataImgTheater",
user_id: int,
uid: int,
) -> List[List[InlineKeyboardButton]]:
floors = [i.round_id for i in abyss_data.abyss_data.detail.rounds_data if i.round_id]
floors.sort()
buttons = [
InlineKeyboardButton(
f"{i}",
callback_data=f"get_role_combat_history|{user_id}|{uid}|{data_id}|{i}",
)
for i in floors
]
send_buttons = [buttons[i : i + 4] for i in range(0, len(buttons), 4)]
all_buttons = [
InlineKeyboardButton(
"<< 返回",
callback_data=f"get_role_combat_history|{user_id}|{uid}|p_1",
),
InlineKeyboardButton(
"总览",
callback_data=f"get_role_combat_history|{user_id}|{uid}|{data_id}|total",
),
InlineKeyboardButton(
"所有",
callback_data=f"get_role_combat_history|{user_id}|{uid}|{data_id}|all",
),
]
send_buttons.append(all_buttons)
return send_buttons
@handler.command("role_combat_history", block=False)
@handler.message(filters.Regex(r"^幻想真境剧诗历史数据"), block=False)
async def abyss_history_command_start(self, update: Update, _: CallbackContext) -> None:
user_id = await self.get_real_user_id(update)
uid, offset = self.get_real_uid_or_offset(update)
message = update.effective_message
self.log_user(update, logger.info, "查询幻想真境剧诗历史数据")
async with self.helper.genshin_or_public(user_id, uid=uid, offset=offset) as client:
await self.get_session_button_data(user_id, client.player_id, force=True)
buttons = await self.gen_season_button(user_id, client.player_id)
if not buttons:
await message.reply_text("还没有幻想真境剧诗历史数据哦~")
return
await message.reply_text("请选择要查询的幻想真境剧诗历史数据", reply_markup=InlineKeyboardMarkup(buttons))
async def get_role_combat_history_page(self, update: "Update", user_id: int, uid: int, result: str):
"""翻页处理"""
callback_query = update.callback_query
self.log_user(update, logger.info, "切换幻想真境剧诗历史数据页 page[%s]", result)
page = int(result.split("_")[1])
async with self.helper.genshin_or_public(user_id, uid=uid) as client:
buttons = await self.gen_season_button(user_id, client.player_id, page)
if not buttons:
await callback_query.answer("还没有幻想真境剧诗历史数据哦~", show_alert=True)
await callback_query.edit_message_text("还没有幻想真境剧诗历史数据哦~")
return
await callback_query.edit_message_reply_markup(reply_markup=InlineKeyboardMarkup(buttons))
await callback_query.answer(f"已切换到第 {page}", show_alert=False)
async def get_role_combat_history_season(self, update: "Update", data_id: int):
"""进入选择幕数"""
callback_query = update.callback_query
user = callback_query.from_user
self.log_user(update, logger.info, "切换幻想真境剧诗历史数据到幕数页 data_id[%s]", data_id)
data = await self.history_data_abyss.get_by_id(data_id)
if not data:
await callback_query.answer("数据不存在,请尝试重新发送命令~", show_alert=True)
await callback_query.edit_message_text("数据不存在,请尝试重新发送命令~")
return
abyss_data = HistoryDataImgTheater.from_data(data)
buttons = await self.gen_floor_button(data_id, abyss_data, user.id, data.user_id)
await callback_query.edit_message_reply_markup(reply_markup=InlineKeyboardMarkup(buttons))
await callback_query.answer("已切换到幕数页", show_alert=False)
async def get_role_combat_history_floor(self, update: "Update", data_id: int, detail: str):
"""渲染幕数数据"""
callback_query = update.callback_query
message = callback_query.message
reply = None
if message.reply_to_message:
reply = message.reply_to_message
floor = 0
total = False
if detail == "total":
floor = 0
elif detail == "all":
total = True
else:
floor = int(detail)
data = await self.history_data_abyss.get_by_id(data_id)
if not data:
await callback_query.answer("数据不存在,请尝试重新发送命令", show_alert=True)
await callback_query.edit_message_text("数据不存在,请尝试重新发送命令~")
return
abyss_data = HistoryDataImgTheater.from_data(data)
images = await self.get_rendered_pic(
abyss_data.abyss_data, abyss_data.character_data, data.user_id, floor, total
)
if images is None:
await callback_query.answer(f"还没有第 {floor} 幕的挑战数据", show_alert=True)
return
await callback_query.answer("正在渲染图片中 请稍等 请不要重复点击按钮", show_alert=False)
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
for group in ArkoWrapper(images).group(10): # 每 10 张图片分一个组
await RenderGroupResult(results=group).reply_media_group(reply or message, write_timeout=60)
self.log_user(update, logger.info, "[bold]幻想真境剧诗挑战数据[/bold]: 成功发送图片", extra={"markup": True})
self.add_delete_message_job(message, delay=1)
@handler.callback_query(pattern=r"^get_role_combat_history\|", block=False)
async def get_role_combat_history(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE") -> None:
callback_query = update.callback_query
user = callback_query.from_user
async def get_role_combat_history_callback(
callback_query_data: str,
) -> Tuple[str, str, int, int]:
_data = callback_query_data.split("|")
_user_id = int(_data[1])
_uid = int(_data[2])
_result = _data[3]
_detail = _data[4] if len(_data) > 4 else None
logger.debug(
"callback_query_data函数返回 detail[%s] result[%s] user_id[%s] uid[%s]",
_detail,
_result,
_user_id,
_uid,
)
return _detail, _result, _user_id, _uid
detail, result, user_id, uid = await get_role_combat_history_callback(callback_query.data)
if user.id != user_id:
await callback_query.answer(text="这不是你的按钮!\n" + config.notice.user_mismatch, show_alert=True)
return
if result == "empty_data":
await callback_query.answer(text="此按钮不可用", show_alert=True)
return
if result.startswith("p_"):
await self.get_role_combat_history_page(update, user_id, uid, result)
return
data_id = int(result)
if detail:
await self.get_role_combat_history_floor(update, data_id, detail)
return
await self.get_role_combat_history_season(update, data_id)
async def abyss_use_by_inline(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE", previous: bool):
callback_query = update.callback_query
user = update.effective_user
user_id = user.id
uid = IInlineUseData.get_uid_from_context(context)
self.log_user(update, logger.info, "查询幻想真境剧诗挑战总览数据 previous[%s]", previous)
notice = None
try:
async with self.helper.genshin_or_public(user_id, uid=uid) as client:
if not client.public:
await client.get_record_cards()
abyss_data, avatar_data = await self.get_rendered_pic_data(client, client.player_id, previous)
images = await self.get_rendered_pic(abyss_data, avatar_data, client.player_id, 0, False)
image = images[0]
except AbyssUnlocked: # 若幻想真境剧诗未解锁
notice = "还未解锁幻想真境剧诗哦~"
except NoMostKills: # 若幻想真境剧诗还未挑战
notice = "还没有挑战本次幻想真境剧诗呢,咕咕咕~"
except AbyssNotFoundError:
notice = "无法查询玩家挑战队伍详情,只能查询统计详情哦~"
except TooManyRequestPublicCookies:
notice = "查询次数太多,请您稍后重试"
if notice:
await callback_query.answer(notice, show_alert=True)
return
await image.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="本期幻想真境剧诗挑战总览",
hash="role_combat_current",
callback=partial(self.abyss_use_by_inline, previous=False),
player=True,
),
IInlineUseData(
text="上期幻想真境剧诗挑战总览",
hash="role_combat_previous",
callback=partial(self.abyss_use_by_inline, previous=True),
player=True,
),
]

View File

@ -1,6 +1,6 @@
import datetime
from asyncio import sleep
from typing import TYPE_CHECKING, List
from typing import TYPE_CHECKING, List, Dict
from simnet.errors import (
TimedOut as SimnetTimedOut,
@ -11,13 +11,18 @@ from telegram.constants import ParseMode
from telegram.error import BadRequest, Forbidden
from core.plugin import Plugin, job
from core.services.history_data.services import HistoryDataAbyssServices, HistoryDataLedgerServices
from core.services.history_data.services import (
HistoryDataAbyssServices,
HistoryDataLedgerServices,
HistoryDataImgTheaterServices,
)
from gram_core.basemodel import RegionEnum
from gram_core.plugin import handler
from gram_core.services.cookies import CookiesService
from gram_core.services.cookies.models import CookiesStatusEnum
from plugins.genshin.abyss import AbyssPlugin
from plugins.genshin.ledger import LedgerPlugin
from plugins.genshin.role_combat import RoleCombatPlugin
from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
from utils.log import logger
@ -43,11 +48,13 @@ class RefreshHistoryJob(Plugin):
genshin_helper: GenshinHelper,
history_abyss: HistoryDataAbyssServices,
history_ledger: HistoryDataLedgerServices,
history_img_theater: HistoryDataImgTheaterServices,
):
self.cookies = cookies
self.genshin_helper = genshin_helper
self.history_data_abyss = history_abyss
self.history_data_ledger = history_ledger
self.history_data_img_theater = history_img_theater
@staticmethod
async def send_notice(context: "ContextTypes.DEFAULT_TYPE", user_id: int, notice_text: str):
@ -58,11 +65,14 @@ class RefreshHistoryJob(Plugin):
except Exception as exc:
logger.error("执行自动刷新历史记录时发生错误 user_id[%s]", user_id, exc_info=exc)
async def save_abyss_data(self, client: "GenshinClient") -> bool:
@staticmethod
async def get_genshin_characters(client: "GenshinClient"):
uid = client.player_id
return await client.get_genshin_characters(uid, lang="zh-cn")
async def save_abyss_data(self, client: "GenshinClient", avatar_data: Dict[int, int]) -> bool:
uid = client.player_id
abyss_data = await client.get_genshin_spiral_abyss(uid, previous=False, lang="zh-cn")
avatars = await client.get_genshin_characters(uid, lang="zh-cn")
avatar_data = {i.id: i.constellation for i in avatars}
if abyss_data.unlocked and abyss_data.ranks and abyss_data.ranks.most_kills:
return await AbyssPlugin.save_abyss_data(self.history_data_abyss, uid, abyss_data, avatar_data)
return False
@ -101,6 +111,20 @@ class RefreshHistoryJob(Plugin):
notice_text = NOTICE_TEXT % ("旅行札记历史记录", now, uid, "旅行札记历史记录")
await self.send_notice(context, user_id, notice_text)
async def save_img_theater_data(self, client: "GenshinClient", avatar_data: Dict[int, int]) -> bool:
uid = client.player_id
abyss_data = await client.get_genshin_imaginarium_theater(uid, previous=False, lang="zh-cn")
if abyss_data.unlocked and abyss_data.data:
data = abyss_data.data[0]
if data.has_data and data.has_detail_data and data.detail:
return await RoleCombatPlugin.save_abyss_data(self.history_data_img_theater, uid, data, avatar_data)
return False
async def send_img_theater_notice(self, context: "ContextTypes.DEFAULT_TYPE", user_id: int, uid: int):
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
notice_text = NOTICE_TEXT % ("幻想真境剧诗历史记录", now, uid, "挑战记录")
await self.send_notice(context, user_id, notice_text)
@handler.command(command="remove_same_history", block=False, admin=True)
async def remove_same_history(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE"):
user = update.effective_user
@ -112,6 +136,8 @@ class RefreshHistoryJob(Plugin):
text += f"深渊数据移除数量:{num1}\n"
num2 = await self.history_data_ledger.remove_same_data()
text += f"旅行札记数据移除数量:{num2}\n"
num3 = await self.history_data_img_theater.remove_same_data()
text += f"幻想真境剧诗数据移除数量:{num3}\n"
await reply.edit_text(text)
@handler.command(command="refresh_all_history", block=False, admin=True)
@ -133,8 +159,12 @@ class RefreshHistoryJob(Plugin):
user_id = cookie_model.user_id
try:
async with self.genshin_helper.genshin(user_id) as client:
if await self.save_abyss_data(client):
await self.send_abyss_notice(context, user_id, client.player_id)
if avatars := await self.get_genshin_characters(client):
avatar_data = {i.id: i.constellation for i in avatars}
if await self.save_abyss_data(client, avatar_data):
await self.send_abyss_notice(context, user_id, client.player_id)
if await self.save_img_theater_data(client, avatar_data):
await self.send_img_theater_notice(context, user_id, client.player_id)
if await self.save_ledger_data(client):
await self.send_ledger_notice(context, user_id, client.player_id)
except (InvalidCookies, PlayerNotFoundError, CookiesNotFoundError):

View File

@ -91,7 +91,7 @@ rich==13.7.1
sentry-sdk==2.11.0
setuptools==71.1.0
shellingham==1.5.4
simnet @ git+https://github.com/PaiGramTeam/SIMNet@d27e6e9e46e48e32ce70abb19a4030a1a93c940c
simnet @ git+https://github.com/PaiGramTeam/SIMNet@0ddc9dabd1543450a7c3f513f177882ed04425fd
six==1.16.0
smmap==5.0.1
sniffio==1.3.1

Binary file not shown.

After

Width:  |  Height:  |  Size: 22 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.3 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.4 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 965 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.9 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 319 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 16 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 15 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.4 KiB

View File

@ -0,0 +1,113 @@
<!DOCTYPE html>
<html lang="zh-cn">
<head>
<meta charset="UTF-8">
<title>floor</title>
<link type="text/css" href="./style.css" rel="stylesheet"/>
<link type="text/css" href="../../styles/public.css" rel="stylesheet"/>
<style>
body {
margin: 0;
padding: 0;
}
.floors, .floor {
border-radius: unset;
margin: 0;
}
.floor-num > div:last-child {
display: flex;
flex-flow: column;
justify-content: center;
align-content: center;
}
</style>
</head>
<body>
<div class="container">
<div class="floors">
<div
class="floor"
style="
background-image: url('./background/floor.png');
background-color: #4d2250;"
>
<div class="head">
<div class="floor-name">
<div class="floor-num"></div>
<div>
<div>UID: {{ uid }}</div>
<div>第 {{ floor.round_id }} 幕</div>
</div>
</div>
{% if floor.is_get_medal %}
<div class="star"></div>
{% endif %}
</div>
<div class="hr"></div>
<div class="chamber">
<div class="chamber-info">
<div>
<span style="color: navajowhite">{{ floor_time }}</span>
</div>
</div>
<div class="battles">
<div class="battle">
{% for character in floor.avatars %}
<div class="character">
{% if character.avatar_type == 1 %}
{% if avatar_data[character.id] > 0 %}
{% set constellation = avatar_data[character.id] %}
{% set bg = ['blue','blue', 'green','green', 'red', 'red'][constellation - 1] %}
<div style="background-color: var(--{{ bg }})">
{{ constellation }} 命
</div>
{% endif %}
{% elif character.avatar_type == 2 %}
<div style="background-color: var(--red)">
试用
</div>
{% elif character.avatar_type == 3 %}
<div style="background-color: var(--green)">
支援
</div>
{% endif %}
<div class="element"
style="background-image: url('../../img/element/{{ character.element }}.png')"></div>
<div class="icon"
style="background-image: url('../../background/rarity/half/{{ character.rarity }}.png')">
<img src="{{ character.icon }}" alt=""/>
</div>
<div class="caption">Lv.{{ character.level }}</div>
</div>
{% endfor %}
</div>
</div>
</div>
<div class="hr"></div>
<div class="chamber">
<h2>奇妙助益 * {{ floor.buffs | length }}</h2>
{% for buff in floor.buffs %}
<div class="buff">
<img class="icon" src="{{ buff.icon }}" alt=""/>
<div class="caption">{{ buff.name }}{{ buff.desc_html | safe }}</div>
</div>
{% endfor %}
</div>
<div class="hr"></div>
<div class="chamber">
<h2>神秘收获 * {{ floor.choice_cards | length }}</h2>
{% for buff in floor.choice_cards %}
<div class="buff">
<img class="icon" src="{{ buff.icon }}" alt=""/>
<div class="caption">{{ buff.name }}{{ buff.desc_html | safe }}</div>
</div>
{% endfor %}
</div>
</div>
</div>
</div>
</body>
</html>

View File

@ -0,0 +1,57 @@
<!DOCTYPE html>
<html lang="zh-ch">
<head>
<meta charset="UTF-8">
<title>overview</title>
<link type="text/css" href="./style.css" rel="stylesheet"/>
<link type="text/css" href="../../styles/public.css" rel="stylesheet"/>
<style>
body {
margin: 0;
padding: 0;
}
.overview {
border-radius: unset;
}
</style>
</head>
<body>
<div class="container">
<div class="overview">
<div class="title">演出回顾</div>
<div class="summarize">
<div>
<div>UID: {{ uid }}</div>
<div>时间: {{ time }}</div>
</div>
<div>
<div class="star">
<span>明星挑战星章 {{ stat.medal_num }} / {{ stat.max_round_id }}</span>
</div>
<div class="star flower">
<span>消耗幻剧之花 {{ stat.coin_num }}</span>
</div>
</div>
<div>
{% set diff = stat.difficulty %}
{% set diff_str = "未知" %}
{% if diff == 1 %}
{% set diff_str = "简单" %}
{% elif diff == 2 %}
{% set diff_str = "普通" %}
{% elif diff == 3 %}
{% set diff_str = "困难" %}
{% endif %}
<div>挑战难度:{{ diff_str }}</div>
<div>最佳记录: 第 {{ stat.max_round_id }} 幕</div>
</div>
<div>
<div>触发场外观众声援: {{ stat.audience_support_trigger_num }}</div>
<div>助演角色支援其他玩家:{{ stat.rent_cnt }}</div>
</div>
</div>
</div>
</div>
</body>
</html>

View File

@ -0,0 +1,376 @@
:root {
--white: rgb(246 248 249);
--bg-color: rgb(233 229 220);
--h-color: rgb(203 189 162);
--red: rgb(255 86 33/ 80%);
--blue: rgb(98 168 233/ 80%);
--green: rgb(67 185 124/ 80%);
}
body {
margin: 0;
padding: 10px;
}
.hr {
width: 100%;
height: 3px;
background-color: rgb(246 248 249 / 50%);
}
.container {
width: 750px;
position: relative;
filter: drop-shadow(2px 2px 5px rgb(0 0 0 /70%));
}
.title {
text-align: center;
font-size: 27px;
font-weight: bold;
color: var(--h-color);
}
.caption {
margin: 10px 0;
color: var(--h-color);
font-size: 20px;
}
/* 概览 */
.overview {
height: 320px;
padding: 20px 40px;
background-size: cover;
background-repeat: no-repeat;
background-position: center;
background-image: linear-gradient(to top, rgb(0 0 0 / 10%), rgb(0 0 0 / 10%)), url("./background/lookback-bg.jpg");
background-attachment: local;
border-radius: 15px;
overflow: hidden;
}
.summarize {
font-size: 20px;
width: calc(100% - 90px);
height: 200px;
margin: 20px 0;
padding: 10px 0 10px 80px;
border-radius: 5px;
border: 2px solid rgb(118 121 120 / 80%);
outline: 4px solid rgb(74, 82, 101);
background-color: rgb(74 82 101 / 60%);
background-image: url("./background/banner 01.png"), url("./background/banner 02.png");
background-repeat: no-repeat, no-repeat;
background-position: right, left;
background-size: auto 100%, auto 100%;
backdrop-filter: blur(5px);
}
.summarize > div {
width: 100%;
height: 25%;
color: var(--white);
display: flex;
align-items: center;
}
.summarize > div > div {
flex: 1;
}
.star {
display: inline-flex;
align-items: center;
position: relative;
}
.star::before {
content: "";
width: 40px;
height: 40px;
background-size: cover;
background-image: url("./background/star.png");
}
.star > span {
margin-left: 10px;
}
.flower::before {
background-image: url("./background/flower.png");
}
.most-played {
width: 100%;
margin: 30px 0;
}
.characters {
width: calc(100% - 60px);
padding: 0 30px;
display: flex;
justify-content: center;
align-items: center;
}
.character {
width: 100px;
height: 125px;
margin: 0 20px;
background-color: rgb(233 229 220);
overflow: hidden;
border-radius: 10px;
position: relative;
}
.characters > .character > .element {
position: absolute;
top: 3px;
left: 3px;
width: 25px;
height: 25px;
background-size: contain;
background-position: center;
background-repeat: no-repeat;
}
.icon {
width: 100%;
height: 100px;
background-size: cover;
background-repeat: no-repeat;
background-position: center;
overflow: hidden;
border-radius: 0 0 20px 0;
display: flex;
flex-direction: column;
flex-wrap: wrap;
align-items: center;
justify-content: space-around;
}
.icon > img {
width: inherit;
height: inherit;
}
.character > .caption {
font-size: 18px;
margin: 4px 0 0;
padding: 0;
height: min-content;
text-align: center;
color: black;
}
.four-star {
width: 15px;
height: 15px;
position: relative;
}
.four-star::after {
content: '';
position: absolute;
top: 0;
left: 0;
z-index: 1;
width: inherit;
height: inherit;
background-position: center;
background-image: url("./background/four-star.png");
background-blend-mode: lighten;
background-size: contain;
}
.ranks {
width: calc(100% + 80px);
transform: translateX(-40px);
color: var(--white);
position: relative;
display: flex;
flex-flow: column;
flex-wrap: wrap;
}
.ranks > div {
font-size: 18px;
width: 100%;
height: 43px;
display: flex;
align-items: center;
}
.rank {
flex: 1;
position: relative;
display: flex;
margin-left: 80px;
margin-right: 60px;
}
.ranks > div > .rank:first-child {
margin-right: 10px;
}
.rank > img {
width: 50px;
position: absolute;
right: 0;
bottom: -5px;
}
/* 层数 */
.floors {
margin-top: 20px;
}
.floor {
margin: 20px 0;
padding: 0 30px;
width: calc(100% - 60px);
border-radius: 20px;
overflow: hidden;
outline: rgb(233 229 220 / 30%) solid 3px;
outline-offset: -7px;
background-position: top center;
background-repeat: no-repeat;
background-size: contain;
}
.head {
width: 100%;
height: 160px;
display: flex;
}
.floor-name {
font-size: 28px;
display: flex;
align-items: center;
font-weight: bold;
color: var(--white);
}
.floor-name > div:not(.floor-num) {
margin-left: 20px;
text-shadow: 1px 1px 5px rgb(0 0 0/50%);
}
.floor-num {
color: black;
width: 100px;
height: 100px;
background-image: url("./background/abyss.png");
background-size: contain;
filter: drop-shadow(0 0 5px rgb(0 0 0/80%)) grayscale(10%);
position: relative;
}
.head > .star {
margin-left: auto;
float: right;
color: var(--white);
font-size: 25px;
text-shadow: 1px 1px 5px rgb(0 0 0/50%);
}
.head > .star::before {
width: 50px;
height: 50px;
}
.chamber {
margin: 10px 0;
display: flex;
flex-flow: column;
}
.chamber-info {
height: 70px;
display: flex;
align-items: center;
color: var(--white);
font-size: 22px;
}
.stars {
margin-left: auto;
}
.dim-star::before {
filter: brightness(50%);
}
.battle {
margin: 10px 0 20px 0;
display: flex;
justify-content: center;
align-items: center;
}
.battle > .character {
flex: 2;
max-width: 120px;
height: 145px;
}
.battle > .battle-info {
flex: 1;
color: var(--white);
font-size: 22px;
font-weight: bold;
}
.battle > .character > div:first-child:not(.icon, .element) {
position: absolute;
top: 0;
right: 0;
padding: 5px;
min-width: 27px;
text-align: center;
border-radius: 0 0 0 10px;
filter: drop-shadow(1px 1px 5px rgb(0 0 0/50%));
font-weight: 500;
color: var(--white);
}
.battle > .character > .element {
position: absolute;
top: 3px;
left: 3px;
width: 30px;
height: 30px;
background-size: contain;
background-position: center;
background-repeat: no-repeat;
}
.battle > .character > .icon {
width: 100%;
height: fit-content;
border-radius: 0 0 28px 0;
}
.battle > .character > .caption {
font-size: 20px;
margin-top: 2px;
}
.chamber h2 {
color: white;
}
.buff {
display: flex;
align-items: center; /* 垂直居中对齐 */
}
.buff img {
width: 100px;
height: auto;
margin-right: 10px; /* 图片与文本之间的间距 */
}