PaiGramTeam/PamGram
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/PamGram.git synced 2024-11-16 03:55:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Support ledger history

Browse Source
This commit is contained in:
xtaodada 2024-05-24 20:31:53 +08:00
parent f4158e0f6e
commit b61a87d0f0
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
4 changed files with 211 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
core/services/history_data/models.py
View File

@ -3,6 +3,7 @@ import enum
from pydantic import BaseModel
from simnet.models.starrail.chronicle.challenge import StarRailChallenge
from simnet.models.starrail.chronicle.challenge_story import StarRailChallengeStory, StarRailChallengeStoryGroup
from simnet.models.starrail.diary import StarRailDiary
from gram_core.services.history_data.models import HistoryData
@ -11,12 +12,14 @@ __all__ = (
"HistoryDataTypeEnum",
"HistoryDataAbyss",
"HistoryDataChallengeStory",
"HistoryDataLedger",
)
class HistoryDataTypeEnum(int, enum.Enum):
ABYSS = 0 # 混沌回忆
CHALLENGE_STORY = 1 # 虚构叙事
LEDGER = 2 # 开拓月历
class HistoryDataAbyss(BaseModel):
@ -34,3 +37,11 @@ class HistoryDataChallengeStory(BaseModel):
@classmethod
def from_data(cls, data: HistoryData) -> "HistoryDataChallengeStory":
return cls.parse_obj(data.data)
class HistoryDataLedger(BaseModel):
diary_data: StarRailDiary
@classmethod
def from_data(cls, data: HistoryData) -> "HistoryDataLedger":
return cls.parse_obj(data.data)

19
core/services/history_data/services.py
View File

@ -4,12 +4,14 @@ from typing import List
from pytz import timezone
from simnet.models.starrail.chronicle.challenge import StarRailChallenge
from simnet.models.starrail.chronicle.challenge_story import StarRailChallengeStory, StarRailChallengeStoryGroup
from simnet.models.starrail.diary import StarRailDiary
from core.services.history_data.models import (
HistoryData,
HistoryDataTypeEnum,
HistoryDataAbyss,
HistoryDataChallengeStory,
HistoryDataLedger,
)
from gram_core.base_service import BaseService
from gram_core.services.history_data.services import HistoryDataBaseServices
@ -24,6 +26,7 @@ __all__ = (
"HistoryDataBaseServices",
"HistoryDataAbyssServices",
"HistoryDataChallengeStoryServices",
"HistoryDataLedgerServices",
)
TZ = timezone("Asia/Shanghai")
@ -75,3 +78,19 @@ class HistoryDataChallengeStoryServices(BaseService, HistoryDataBaseServices):
type=HistoryDataChallengeStoryServices.DATA_TYPE,
data=dict_data,
)
class HistoryDataLedgerServices(BaseService, HistoryDataBaseServices):
DATA_TYPE = HistoryDataTypeEnum.LEDGER.value
@staticmethod
def create(user_id: int, diary_data: StarRailDiary):
data = HistoryDataLedger(diary_data=diary_data)
json_data = data.json(by_alias=True, encoder=json_encoder)
return HistoryData(
user_id=user_id,
data_id=diary_data.data_id,
time_created=datetime.datetime.now(),
type=HistoryDataLedgerServices.DATA_TYPE,
data=jsonlib.loads(json_data),
)

1
plugins/admin/set_command.py
View File

@ -35,6 +35,7 @@ class SetCommandPlugin(Plugin):
BotCommand("dailynote", "查询实时便笺"),
BotCommand("redeem", "（国际服）兑换 Key"),
BotCommand("ledger", "查询当月开拓月历"),
BotCommand("ledger_history", "查询开拓月历历史记录"),
BotCommand("avatars", "查询角色练度"),
BotCommand("player_card", "角色卡片"),
BotCommand("role_detail", "角色详细信息"),

185
plugins/starrail/ledger.py
View File

@ -1,24 +1,31 @@
import math
import os
import re
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, List, Tuple
from simnet.errors import BadRequest as SimnetBadRequest, DataNotPublic
from simnet.models.starrail.diary import StarRailDiary
from telegram import Update
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ChatAction
from telegram.ext import filters, CallbackContext
from telegram.ext import filters, CallbackContext, ContextTypes
from core.plugin import Plugin, handler
from core.services.cookies import CookiesService
from core.services.history_data.models import HistoryDataLedger
from core.services.history_data.services import HistoryDataLedgerServices
from core.services.template.models import RenderResult
from core.services.template.services import TemplateService
from gram_core.config import config
from gram_core.dependence.redisdb import RedisDB
from plugins.tools.genshin import GenshinHelper
from utils.enkanetwork import RedisCache
from utils.log import logger
from utils.uid import mask_number
if TYPE_CHECKING:
from simnet import StarRailClient
from simnet.models.starrail.diary import StarRailDiary
__all__ = ("LedgerPlugin",)
@ -32,15 +39,24 @@ class LedgerPlugin(Plugin):
helper: GenshinHelper,
cookies_service: CookiesService,
template_service: TemplateService,
history_data_ledger: HistoryDataLedgerServices,
redis: RedisDB,
):
self.template_service = template_service
self.cookies_service = cookies_service
self.current_dir = os.getcwd()
self.helper = helper
self.history_data_ledger = history_data_ledger
self.cache = RedisCache(redis.client, key="plugin:ledger:history")
self.kitsune = None
async def _start_get_ledger(self, client: "StarRailClient", year, month) -> RenderResult:
req_month = f"{year}0{month}" if month < 10 else f"{year}{month}"
diary_info: StarRailDiary = await client.get_starrail_diary(client.player_id, month=req_month)
await self.save_ledger_data(client.player_id, diary_info)
return await self._start_get_ledger_render(client.player_id, diary_info)
async def _start_get_ledger_render(self, uid: int, diary_info: "StarRailDiary") -> RenderResult:
color = ["#73a9c6", "#d56565", "#70b2b4", "#bd9a5a", "#739970", "#7a6da7", "#597ea0"]
categories = [
{
@ -58,8 +74,8 @@ class LedgerPlugin(Plugin):
return f"{round(amount / 10000, 2)}w" if amount >= 10000 else amount
ledger_data = {
"uid": mask_number(client.player_id),
"day": month,
"uid": mask_number(uid),
"day": diary_info.month,
"current_hcoin": format_amount(diary_info.month_data.current_hcoin),
"gacha": int(diary_info.month_data.current_hcoin / 160),
"current_rails_pass": format_amount(diary_info.month_data.current_rails_pass),
@ -131,3 +147,162 @@ class LedgerPlugin(Plugin):
raise exc
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename=f"{client.player_id}.png", allow_sending_without_reply=True)
async def save_ledger_data(self, uid: int, ledger_data: "StarRailDiary"):
if int(ledger_data.current_month) == ledger_data.month:
return
model = self.history_data_ledger.create(uid, ledger_data)
old_data = await self.history_data_ledger.get_by_user_id_data_id(uid, model.data_id)
if not old_data:
await self.history_data_ledger.add(model)
async def get_ledger_data(self, uid: int):
return await self.history_data_ledger.get_by_user_id(uid)
@staticmethod
def get_season_data_name(data: "HistoryDataLedger") -> str:
return f"{data.diary_data.data_id}"
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_ledger_data(uid)
data.sort(key=lambda x: x.data_id, reverse=True)
abyss_data = [HistoryDataLedger.from_data(i) for i in data]
buttons = [
{
"name": LedgerPlugin.get_season_data_name(abyss_data[idx]),
"value": f"get_ledger_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_ledger_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_ledger_history|{user_id}|{uid}|empty_data",
)
)
if next_page:
last_button.append(
InlineKeyboardButton(
"下一页 >>",
callback_data=f"get_ledger_history|{user_id}|{uid}|p_{next_page}",
)
)
if last_button:
send_buttons.append(last_button)
return send_buttons
@handler.command("ledger_history", cookie=True, block=False)
@handler.message(filters.Regex(r"^开拓月历历史数据"), block=False)
async def ledger_history_command_start(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE") -> None:
user_id = await self.get_real_user_id(update)
message = update.effective_message
self.log_user(update, logger.info, "查询开拓月历历史数据")
async with self.helper.genshin(user_id) 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
if isinstance(self.kitsune, str):
photo = self.kitsune
else:
photo = open("resources/img/aaa.jpg", "rb")
reply_message = await message.reply_photo(
photo, "请选择要查询的开拓月历历史数据", reply_markup=InlineKeyboardMarkup(buttons)
)
if reply_message.photo:
self.kitsune = reply_message.photo[-1].file_id
async def get_ledger_history_page(self, update: "Update", user_id: 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(user_id) 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)
@handler.callback_query(pattern=r"^get_ledger_history\|", block=False)
async def get_ledger_history(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE") -> None:
callback_query = update.callback_query
message = callback_query.message
user = callback_query.from_user
async def get_ledger_history_callback(
callback_query_data: str,
) -> Tuple[str, int, int]:
_data = callback_query_data.split("|")
_user_id = int(_data[1])
_uid = int(_data[2])
_result = _data[3]
logger.debug(
"callback_query_data函数返回 result[%s] user_id[%s] uid[%s]",
_result,
_user_id,
_uid,
)
return _result, _user_id, _uid
result, user_id, _ = await get_ledger_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_ledger_history_page(update, user_id, result)
return
data_id = int(result)
data = await self.history_data_ledger.get_by_id(data_id)
if not data:
await callback_query.answer("数据不存在，请尝试重新发送命令", show_alert=True)
await callback_query.edit_message_text("数据不存在，请尝试重新发送命令~")
return
await callback_query.answer("正在渲染图片中 请稍等 请不要重复点击按钮")
render = await self._start_get_ledger_render(user_id, HistoryDataLedger.from_data(data).diary_data)
await render.edit_media(message)