Support inline use

This commit is contained in:
omg-xtao 2024-06-19 16:29:47 +08:00 committed by GitHub
parent 519c46e619
commit c03304d101
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 487 additions and 46 deletions

View File

@ -32,6 +32,8 @@ OWNER=0
# 文章推送群组 可选配置项 # 文章推送群组 可选配置项
# CHANNELS=[] # CHANNELS=[]
# 消息帮助频道 可选配置项
# CHANNELS_HELPER=0
# 是否允许机器人邀请到其他群 默认不允许 如果允许 可以允许全部人或有认证选项 可选配置项 # 是否允许机器人邀请到其他群 默认不允许 如果允许 可以允许全部人或有认证选项 可选配置项
# JOIN_GROUPS = "NO_ALLOW" # JOIN_GROUPS = "NO_ALLOW"

@ -1 +1 @@
Subproject commit c478924e064d9dff9a944107ce34ee639f3febf6 Subproject commit 9bce0f921f7400701edcf5191b97ce89e58007db

View File

@ -1,23 +1,30 @@
import asyncio import asyncio
from typing import Awaitable, Dict, List, cast from typing import Awaitable, Dict, List, cast, Tuple
from uuid import uuid4 from uuid import uuid4
from telegram import ( from telegram import (
InlineQuery, InlineQuery,
InlineQueryResultArticle, InlineQueryResultArticle,
InlineQueryResultPhoto,
InlineQueryResultCachedPhoto, InlineQueryResultCachedPhoto,
InputTextMessageContent, InputTextMessageContent,
Update, Update,
InlineQueryResultsButton, InlineQueryResultsButton,
InlineKeyboardButton,
InlineKeyboardMarkup,
) )
from telegram.constants import ParseMode from telegram.constants import ParseMode
from telegram.error import BadRequest from telegram.error import BadRequest
from telegram.ext import CallbackContext from telegram.ext import CallbackContext, ContextTypes
from core.dependence.assets import AssetsCouldNotFound, AssetsService from core.dependence.assets import AssetsCouldNotFound, AssetsService
from core.plugin import Plugin, handler from core.plugin import Plugin, handler
from core.services.cookies import CookiesService
from core.services.players import PlayersService
from core.services.search.services import SearchServices from core.services.search.services import SearchServices
from core.services.wiki.services import WikiService from core.services.wiki.services import WikiService
from gram_core.config import config
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from utils.log import logger from utils.log import logger
@ -29,6 +36,8 @@ class Inline(Plugin):
wiki_service: WikiService, wiki_service: WikiService,
assets_service: AssetsService, assets_service: AssetsService,
search_service: SearchServices, search_service: SearchServices,
cookies_service: CookiesService,
players_service: PlayersService,
): ):
self.assets_service = assets_service self.assets_service = assets_service
self.wiki_service = wiki_service self.wiki_service = wiki_service
@ -36,6 +45,11 @@ class Inline(Plugin):
self.characters_list: List[Dict[str, str]] = [] self.characters_list: List[Dict[str, str]] = []
self.refresh_task: List[Awaitable] = [] self.refresh_task: List[Awaitable] = []
self.search_service = search_service self.search_service = search_service
self.cookies_service = cookies_service
self.players_service = players_service
self.inline_use_data: List[IInlineUseData] = []
self.inline_use_data_map: Dict[str, IInlineUseData] = {}
self.img_url = "https://i.dawnlab.me/b1bdf9cc3061d254f038e557557694bc.jpg"
async def initialize(self): async def initialize(self):
# todo: 整合进 wiki 或者单独模块 从Redis中读取 # todo: 整合进 wiki 或者单独模块 从Redis中读取
@ -72,8 +86,113 @@ class Inline(Plugin):
self.refresh_task.append(asyncio.create_task(task_weapons())) self.refresh_task.append(asyncio.create_task(task_weapons()))
self.refresh_task.append(asyncio.create_task(task_characters())) self.refresh_task.append(asyncio.create_task(task_characters()))
async def init_inline_use_data(self):
if self.inline_use_data:
return
for _, instance in self.application.managers.plugins_map.items():
if _data := await instance.get_inline_use_data():
self.inline_use_data.extend(_data)
for data in self.inline_use_data:
self.inline_use_data_map[data.hash] = data
async def user_base_data(self, user_id: int, player_id: int, offset: int) -> Tuple[int, bool, bool]:
uid, has_cookie, has_player = 0, False, False
player = await self.players_service.get_player(user_id, None, player_id, offset)
if player is not None:
uid = player.player_id
has_player = True
if player.account_id is not None:
cookie_model = await self.cookies_service.get(player.user_id, player.account_id, player.region)
if cookie_model is not None:
has_cookie = True
return uid, has_cookie, has_player
def get_inline_use_button_data(self, user_id: int, uid: int, cookie: bool, player: bool) -> InlineKeyboardMarkup:
button_data = []
start = f"use_inline_func|{user_id}|{uid}"
for data in self.inline_use_data:
if data.is_show(cookie, player):
button_data.append(
InlineKeyboardButton(text=data.text, callback_data=data.get_button_callback_data(start))
)
# 每三个一行
button_data = [button_data[i : i + 3] for i in range(0, len(button_data), 3)]
return InlineKeyboardMarkup(button_data)
@handler.callback_query(pattern=r"^use_inline_func\|", block=False)
async def use_by_inline_query_callback(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE") -> None:
user = update.effective_user
callback_query = update.callback_query
async def get_inline_query_callback(callback_query_data: str) -> Tuple[int, int, str]:
_data = callback_query_data.split("|")
_user_id = int(_data[1])
_uid = int(_data[2])
_hash = _data[3]
logger.debug("callback_query_data函数返回 user_id[%s] uid[%s] hash[%s]", _user_id, _uid, _hash)
return _user_id, _uid, _hash
user_id, uid, hash_str = await get_inline_query_callback(callback_query.data)
if user.id != user_id:
await callback_query.answer(text="这不是你的按钮!\n" + config.notice.user_mismatch, show_alert=True)
return
callback = self.inline_use_data_map.get(hash_str)
if callback is None:
await callback_query.answer(text="数据不存在,请重新生成按钮", show_alert=True)
return
IInlineUseData.set_uid_to_context(context, uid)
await callback.callback(update, context)
@handler.inline_query(pattern="^功能", block=False)
async def use_by_inline_query(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE") -> None:
if not config.channels_helper:
logger.warning("未设置 helper 频道")
return
await self.init_inline_use_data()
user = update.effective_user
ilq = cast(InlineQuery, update.inline_query)
query = ilq.query
switch_pm_text = "需要帮助嘛?"
logger.info("用户 %s[%s] inline_query 功能查询\nquery[%s]", user.full_name, user.id, query)
user_id = user.id
uid, offset = self.get_real_uid_or_offset(update)
real_uid, has_cookie, has_player = await self.user_base_data(user_id, uid, offset)
button_data = self.get_inline_use_button_data(user_id, real_uid, has_cookie, has_player)
try:
await ilq.answer(
results=[
InlineQueryResultPhoto(
id=str(uuid4()),
photo_url=self.img_url,
thumbnail_url=self.img_url,
caption="请从下方按钮选择功能",
reply_markup=button_data,
)
],
cache_time=0,
auto_pagination=True,
button=InlineQueryResultsButton(
text=switch_pm_text,
start_parameter="inline_message",
),
)
except BadRequest as exc:
if "Query is too old" in exc.message: # 过时请求全部忽略
logger.warning("用户 %s[%s] inline_query 请求过时", user.full_name, user.id)
return
if "can't parse entities" not in exc.message:
raise exc
logger.warning("inline_query发生BadRequest错误", exc_info=exc)
await ilq.answer(
results=[],
button=InlineQueryResultsButton(
text="糟糕,发生错误了。",
start_parameter="inline_message",
),
)
@handler.inline_query(block=False) @handler.inline_query(block=False)
async def inline_query(self, update: Update, _: CallbackContext) -> None: async def z_inline_query(self, update: Update, _: CallbackContext) -> None:
user = update.effective_user user = update.effective_user
ilq = cast(InlineQuery, update.inline_query) ilq = cast(InlineQuery, update.inline_query)
query = ilq.query query = ilq.query
@ -98,8 +217,18 @@ class Inline(Plugin):
input_message_content=InputTextMessageContent("角色攻略查询"), input_message_content=InputTextMessageContent("角色攻略查询"),
) )
) )
results_list.append(
InlineQueryResultArticle(
id=str(uuid4()),
title="使用功能",
description="输入 功能 即可直接使用 BOT 功能",
input_message_content=InputTextMessageContent("Inline 模式下输入 功能 即可直接使用 BOT 功能"),
)
)
elif args[0] == "cookies_export": elif args[0] == "cookies_export":
return return
elif args[0] == "功能":
return
else: else:
if args[0] == "查看武器列表并查询": if args[0] == "查看武器列表并查询":
for weapon in self.weapons_list: for weapon in self.weapons_list:

View File

@ -4,7 +4,7 @@ import asyncio
import math import math
import re import re
from datetime import datetime from datetime import datetime
from functools import lru_cache from functools import lru_cache, partial
from typing import Any, Coroutine, List, Optional, Tuple, Union, Dict from typing import Any, Coroutine, List, Optional, Tuple, Union, Dict
from arkowrapper import ArkoWrapper from arkowrapper import ArkoWrapper
@ -24,6 +24,7 @@ from core.services.template.models import RenderGroupResult, RenderResult
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.config import config from gram_core.config import config
from gram_core.dependence.redisdb import RedisDB from gram_core.dependence.redisdb import RedisDB
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from plugins.tools.genshin import GenshinHelper from plugins.tools.genshin import GenshinHelper
from utils.enkanetwork import RedisCache from utils.enkanetwork import RedisCache
from utils.log import logger from utils.log import logger
@ -578,3 +579,49 @@ class AbyssPlugin(Plugin):
await self.get_abyss_history_floor(update, data_id, detail) await self.get_abyss_history_floor(update, data_id, detail)
return return
await self.get_abyss_history_season(update, data_id) await self.get_abyss_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="abyss_current",
callback=partial(self.abyss_use_by_inline, previous=False),
player=True,
),
IInlineUseData(
text="上期深渊挑战总览",
hash="abyss_previous",
callback=partial(self.abyss_use_by_inline, previous=True),
player=True,
),
]

View File

@ -18,6 +18,7 @@ from core.services.players import PlayersService
from core.services.players.services import PlayerInfoService from core.services.players.services import PlayerInfoService
from core.services.template.models import FileType from core.services.template.models import FileType
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from gram_core.services.template.models import RenderGroupResult from gram_core.services.template.models import RenderGroupResult
from modules.wiki.base import Model from modules.wiki.base import Model
from plugins.tools.genshin import CharacterDetails, GenshinHelper from plugins.tools.genshin import CharacterDetails, GenshinHelper
@ -33,6 +34,10 @@ if TYPE_CHECKING:
MAX_AVATAR_COUNT = 40 MAX_AVATAR_COUNT = 40
class TooManyRequests(Exception):
"""请求过多"""
class SkillData(Model): class SkillData(Model):
"""天赋数据""" """天赋数据"""
@ -170,6 +175,29 @@ class AvatarListPlugin(Plugin):
tasks = [render_task(i * image_count, c) for i, c in enumerate(avatar_datas_group)] tasks = [render_task(i * image_count, c) for i, c in enumerate(avatar_datas_group)]
return await asyncio.gather(*tasks) return await asyncio.gather(*tasks)
async def render(
self, client: "GenshinClient", user_id: int, user_name: str, all_avatars: bool = False
) -> List["RenderResult"]:
characters = await client.get_genshin_characters(client.player_id)
avatar_datas: List[AvatarData] = await self.get_avatars_data(
characters, client, None if all_avatars else MAX_AVATAR_COUNT
)
if not avatar_datas:
raise TooManyRequests()
name_card, avatar, nickname, rarity = await self.player_info_system.get_player_info(
client.player_id, user_id, user_name
)
base_render_data = {
"uid": mask_number(client.player_id), # 玩家uid
"nickname": nickname, # 玩家昵称
"avatar": avatar, # 玩家头像
"rarity": rarity, # 玩家头像对应的角色星级
"namecard": name_card, # 玩家名片
"has_more": len(characters) != len(avatar_datas), # 是否显示了全部角色
}
return await self.avatar_list_render(base_render_data, avatar_datas, not all_avatars)
@handler.command("avatars", cookie=True, block=False) @handler.command("avatars", cookie=True, block=False)
@handler.message(filters.Regex(r"^(全部)?练度统计$"), cookie=True, block=False) @handler.message(filters.Regex(r"^(全部)?练度统计$"), cookie=True, block=False)
async def avatar_list(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE"): async def avatar_list(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE"):
@ -180,20 +208,16 @@ class AvatarListPlugin(Plugin):
all_avatars = "全部" in message.text or "all" in message.text # 是否发送全部角色 all_avatars = "全部" in message.text or "all" in message.text # 是否发送全部角色
self.log_user(update, logger.info, "[bold]练度统计[/bold]: all=%s", all_avatars, extra={"markup": True}) self.log_user(update, logger.info, "[bold]练度统计[/bold]: all=%s", all_avatars, extra={"markup": True})
notice = None
try: try:
async with self.helper.genshin(user_id, player_id=uid, offset=offset) as client: async with self.helper.genshin(user_id, player_id=uid, offset=offset) as client:
notice = await message.reply_text(f"{config.notice.bot_name}需要收集整理数据,还请耐心等待哦~") notice = await message.reply_text(f"{config.notice.bot_name}需要收集整理数据,还请耐心等待哦~")
self.add_delete_message_job(notice, delay=60) self.add_delete_message_job(notice, delay=60)
await message.reply_chat_action(ChatAction.TYPING) await message.reply_chat_action(ChatAction.TYPING)
characters = await client.get_genshin_characters(client.player_id) images = await self.render(client, user_id, user_name, all_avatars)
avatar_datas: List[AvatarData] = await self.get_avatars_data( except TooManyRequests:
characters, client, None if all_avatars else MAX_AVATAR_COUNT reply_message = await message.reply_html("服务器熟啦 ~ 请稍后再试")
) self.add_delete_message_job(reply_message, delay=20)
if not avatar_datas: return
reply_message = await message.reply_html("服务器熟啦 ~ 请稍后再试")
self.add_delete_message_job(reply_message, delay=20)
return
except SimnetBadRequest as e: except SimnetBadRequest as e:
if e.ret_code == -502002: if e.ret_code == -502002:
reply_message = await message.reply_html("请先在米游社中使用一次<b>养成计算器</b>后再使用此功能~") reply_message = await message.reply_html("请先在米游社中使用一次<b>养成计算器</b>后再使用此功能~")
@ -201,21 +225,7 @@ class AvatarListPlugin(Plugin):
return return
raise e raise e
name_card, avatar, nickname, rarity = await self.player_info_system.get_player_info(
client.player_id, user_id, user_name
)
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
base_render_data = {
"uid": mask_number(client.player_id), # 玩家uid
"nickname": nickname, # 玩家昵称
"avatar": avatar, # 玩家头像
"rarity": rarity, # 玩家头像对应的角色星级
"namecard": name_card, # 玩家名片
"has_more": len(characters) != len(avatar_datas), # 是否显示了全部角色
}
images = await self.avatar_list_render(base_render_data, avatar_datas, not all_avatars)
for group in ArkoWrapper(images).group(10): # 每 10 张图片分一个组 for group in ArkoWrapper(images).group(10): # 每 10 张图片分一个组
await RenderGroupResult(results=group).reply_media_group(message, write_timeout=60) await RenderGroupResult(results=group).reply_media_group(message, write_timeout=60)
@ -226,3 +236,35 @@ class AvatarListPlugin(Plugin):
"[bold]练度统计[/bold]发送图片成功", "[bold]练度统计[/bold]发送图片成功",
extra={"markup": True}, extra={"markup": True},
) )
async def avatar_list_use_by_inline(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE") -> None:
callback_query = update.callback_query
user_id = await self.get_real_user_id(update)
user_name = self.get_real_user_name(update)
uid = IInlineUseData.get_uid_from_context(context)
self.log_user(update, logger.info, "查询练度统计")
try:
async with self.helper.genshin(user_id, player_id=uid) as client:
images = await self.render(client, user_id, user_name)
render = images[0]
except TooManyRequests:
await callback_query.answer("服务器熟啦 ~ 请稍后再试", show_alert=True)
return
except SimnetBadRequest as e:
if e.ret_code == -502002:
await callback_query.answer("请先在米游社中使用一次养成计算器后再使用此功能~", show_alert=True)
return
raise e
await render.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="练度统计",
hash="avatar_list",
callback=self.avatar_list_use_by_inline,
cookie=True,
player=True,
)
]

View File

@ -1,5 +1,6 @@
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Dict from functools import partial
from typing import Dict, List, Optional, TYPE_CHECKING
from telegram import Update from telegram import Update
from telegram.constants import ChatAction from telegram.constants import ChatAction
@ -9,6 +10,7 @@ from core.dependence.assets import AssetsService
from core.dependence.redisdb import RedisDB from core.dependence.redisdb import RedisDB
from core.plugin import Plugin, handler from core.plugin import Plugin, handler
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from modules.apihelper.client.components.calendar import Calendar from modules.apihelper.client.components.calendar import Calendar
from utils.log import logger from utils.log import logger
@ -17,6 +19,10 @@ try:
except ImportError: except ImportError:
import json as jsonlib import json as jsonlib
if TYPE_CHECKING:
from telegram.ext import ContextTypes
from gram_core.services.template.models import RenderResult
class CalendarPlugin(Plugin): class CalendarPlugin(Plugin):
"""活动日历查询""" """活动日历查询"""
@ -41,6 +47,15 @@ class CalendarPlugin(Plugin):
await self.cache.set("plugin:calendar", jsonlib.dumps(data, default=lambda x: x.dict()), ex=next_hour - now) await self.cache.set("plugin:calendar", jsonlib.dumps(data, default=lambda x: x.dict()), ex=next_hour - now)
return data return data
async def render(self, list_mode: bool) -> "RenderResult":
data = await self._fetch_data()
data["display_mode"] = "list" if list_mode else "calendar"
return await self.template_service.render(
"genshin/calendar/calendar.jinja2",
data,
query_selector=".container",
)
@handler.command("calendar", block=False) @handler.command("calendar", block=False)
@handler(MessageHandler, filters=filters.Regex(r"^(活动)+(日历|日历列表)$"), block=False) @handler(MessageHandler, filters=filters.Regex(r"^(活动)+(日历|日历列表)$"), block=False)
async def command_start(self, update: Update, _: CallbackContext) -> None: async def command_start(self, update: Update, _: CallbackContext) -> None:
@ -48,12 +63,28 @@ class CalendarPlugin(Plugin):
mode = "list" if "列表" in message.text else "calendar" mode = "list" if "列表" in message.text else "calendar"
self.log_user(update, logger.info, "查询日历 | 模式 %s", mode) self.log_user(update, logger.info, "查询日历 | 模式 %s", mode)
await message.reply_chat_action(ChatAction.TYPING) await message.reply_chat_action(ChatAction.TYPING)
data = await self._fetch_data() image = await self.render(mode == "list")
data["display_mode"] = mode
image = await self.template_service.render(
"genshin/calendar/calendar.jinja2",
data,
query_selector=".container",
)
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await image.reply_photo(message) await image.reply_photo(message)
async def calendar_use_by_inline(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE", list_mode: bool):
callback_query = update.callback_query
self.log_user(update, logger.info, "查询日历 | 列表模式 %s", list_mode)
await callback_query.answer("正在查询日历,请耐心等待")
image = await self.render(list_mode)
await image.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="活动日历",
hash="calendar",
callback=partial(self.calendar_use_by_inline, list_mode=False),
),
IInlineUseData(
text="活动日历列表",
hash="calendar_list",
callback=partial(self.calendar_use_by_inline, list_mode=True),
),
]

View File

@ -1,5 +1,5 @@
from datetime import datetime from datetime import datetime
from typing import Optional, TYPE_CHECKING from typing import Optional, TYPE_CHECKING, List
from simnet.errors import DataNotPublic from simnet.errors import DataNotPublic
from telegram import InlineKeyboardMarkup, InlineKeyboardButton from telegram import InlineKeyboardMarkup, InlineKeyboardButton
@ -10,6 +10,7 @@ from telegram.helpers import create_deep_linked_url
from core.plugin import Plugin, handler from core.plugin import Plugin, handler
from core.services.template.models import RenderResult from core.services.template.models import RenderResult
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from plugins.tools.genshin import GenshinHelper from plugins.tools.genshin import GenshinHelper
from utils.log import logger from utils.log import logger
from utils.uid import mask_number from utils.uid import mask_number
@ -127,3 +128,33 @@ class DailyNotePlugin(Plugin):
filename=f"{client.player_id}.png", filename=f"{client.player_id}.png",
reply_markup=self.get_task_button(context.bot.username), reply_markup=self.get_task_button(context.bot.username),
) )
async def daily_note_use_by_inline(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"):
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, "每日便签命令请求")
try:
async with self.helper.genshin(user_id, player_id=uid) as client:
render_result = await self._get_daily_note(client)
except DataNotPublic:
await callback_query.answer(
"查询失败惹,可能是便签功能被禁用了?请尝试通过米游社或者 hoyolab 获取一次便签信息后重试。",
show_alert=True,
)
return ConversationHandler.END
await render_result.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="当前状态",
hash="dailynote",
callback=self.daily_note_use_by_inline,
cookie=True,
player=True,
)
]

View File

@ -1,13 +1,19 @@
from typing import List, Optional, TYPE_CHECKING
from telegram import Update from telegram import Update
from telegram.constants import ChatAction from telegram.constants import ChatAction
from telegram.ext import CallbackContext, filters from telegram.ext import CallbackContext, filters
from core.plugin import Plugin, handler from core.plugin import Plugin, handler
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from utils.log import logger from utils.log import logger
__all__ = ("HelpPlugin",) __all__ = ("HelpPlugin",)
if TYPE_CHECKING:
from gram_core.services.template.models import RenderResult
class HelpPlugin(Plugin): class HelpPlugin(Plugin):
def __init__(self, template_service: TemplateService = None): def __init__(self, template_service: TemplateService = None):
@ -15,17 +21,35 @@ class HelpPlugin(Plugin):
raise ModuleNotFoundError raise ModuleNotFoundError
self.template_service = template_service self.template_service = template_service
async def get_help_render(self) -> "RenderResult":
return await self.template_service.render(
"bot/help/help.jinja2",
{"bot_username": self.application.bot.username},
{"width": 1280, "height": 900},
ttl=30 * 24 * 60 * 60,
)
@handler.command(command="help", block=False) @handler.command(command="help", block=False)
@handler.command(command="start", filters=filters.Regex("inline_message$"), block=False) @handler.command(command="start", filters=filters.Regex("inline_message$"), block=False)
async def start(self, update: Update, _: CallbackContext): async def start(self, update: Update, _: CallbackContext):
message = update.effective_message message = update.effective_message
self.log_user(update, logger.info, "发出help命令") self.log_user(update, logger.info, "发出help命令")
await message.reply_chat_action(ChatAction.TYPING) await message.reply_chat_action(ChatAction.TYPING)
render_result = await self.template_service.render( render_result = await self.get_help_render()
"bot/help/help.jinja2",
{"bot_username": self.application.bot.username},
{"width": 1280, "height": 900},
ttl=30 * 24 * 60 * 60,
)
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await render_result.reply_photo(message, filename="help.png") await render_result.reply_photo(message, filename="help.png")
async def start_use_by_inline(self, update: Update, _: CallbackContext):
callback_query = update.callback_query
self.log_user(update, logger.info, "发出help命令")
render_result = await self.get_help_render()
await render_result.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="帮助",
hash="help",
callback=self.start_use_by_inline,
)
]

View File

@ -2,7 +2,7 @@ import math
import os import os
import re import re
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import TYPE_CHECKING, List, Tuple from typing import TYPE_CHECKING, List, Tuple, Optional
from simnet.errors import DataNotPublic, BadRequest as SimnetBadRequest from simnet.errors import DataNotPublic, BadRequest as SimnetBadRequest
from telegram import InlineKeyboardMarkup, InlineKeyboardButton from telegram import InlineKeyboardMarkup, InlineKeyboardButton
@ -17,6 +17,7 @@ from core.services.template.models import RenderResult
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.config import config from gram_core.config import config
from gram_core.dependence.redisdb import RedisDB from gram_core.dependence.redisdb import RedisDB
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from plugins.tools.genshin import GenshinHelper from plugins.tools.genshin import GenshinHelper
from utils.enkanetwork import RedisCache from utils.enkanetwork import RedisCache
from utils.log import logger from utils.log import logger
@ -312,3 +313,41 @@ class LedgerPlugin(Plugin):
await callback_query.answer("正在渲染图片中 请稍等 请不要重复点击按钮") await callback_query.answer("正在渲染图片中 请稍等 请不要重复点击按钮")
render = await self._start_get_ledger_render(user_id, HistoryDataLedger.from_data(data).diary_data) render = await self._start_get_ledger_render(user_id, HistoryDataLedger.from_data(data).diary_data)
await render.edit_media(message) await render.edit_media(message)
async def ledger_use_by_inline(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"):
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, "查询旅行札记")
try:
async with self.helper.genshin(user_id, player_id=uid) as client:
render_result = await self._start_get_ledger(client)
except DataNotPublic:
await callback_query.answer(
"查询失败惹,可能是旅行札记功能被禁用了?请先通过米游社或者 hoyolab 获取一次旅行札记后重试。",
show_alert=True,
)
return
except SimnetBadRequest as exc:
if exc.ret_code == -120:
await callback_query.answer(
"当前角色冒险等阶不足,暂时无法获取信息",
show_alert=True,
)
return
raise exc
await render_result.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="当月旅行札记",
hash="ledger",
callback=self.ledger_use_by_inline,
cookie=True,
player=True,
)
]

View File

@ -1,5 +1,5 @@
import random import random
from typing import Optional, TYPE_CHECKING from typing import Optional, TYPE_CHECKING, List
from telegram.constants import ChatAction from telegram.constants import ChatAction
from telegram.ext import filters from telegram.ext import filters
@ -8,6 +8,7 @@ from core.plugin import Plugin, handler
from core.services.cookies.error import TooManyRequestPublicCookies from core.services.cookies.error import TooManyRequestPublicCookies
from core.services.template.models import RenderResult from core.services.template.models import RenderResult
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from plugins.tools.genshin import GenshinHelper from plugins.tools.genshin import GenshinHelper
from utils.log import logger from utils.log import logger
from utils.uid import mask_number from utils.uid import mask_number
@ -110,3 +111,41 @@ class PlayerStatsPlugins(Plugin):
item.__config__.allow_mutation = True item.__config__.allow_mutation = True
item.icon = await self.download_resource(item.icon) item.icon = await self.download_resource(item.icon)
item.cover = await self.download_resource(item.cover) item.cover = await self.download_resource(item.cover)
async def stats_use_by_inline(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"):
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, "查询游戏用户命令请求")
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()
render_result = await self.render(client, client.player_id)
except TooManyRequestPublicCookies:
notice = "用户查询次数过多 请稍后重试"
except AttributeError as exc:
logger.error("角色数据有误")
logger.exception(exc)
notice = f"角色数据有误 估计是{config.notice.bot_name}晕了"
except ValueError as exc:
logger.warning("获取 uid 发生错误! 错误信息为 %s", str(exc))
notice = "UID 内部错误"
if notice:
await callback_query.answer(notice, show_alert=True)
return
await render_result.edit_inline_media(callback_query)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
return [
IInlineUseData(
text="玩家统计",
hash="stats",
callback=self.stats_use_by_inline,
player=True,
),
]

View File

@ -1,3 +1,4 @@
from functools import partial
from io import BytesIO from io import BytesIO
from typing import Optional, TYPE_CHECKING, List, Union, Tuple from typing import Optional, TYPE_CHECKING, List, Union, Tuple
from urllib.parse import urlencode from urllib.parse import urlencode
@ -19,6 +20,7 @@ from core.services.players import PlayersService
from core.services.template.models import FileType from core.services.template.models import FileType
from core.services.template.services import TemplateService from core.services.template.services import TemplateService
from gram_core.config import config from gram_core.config import config
from gram_core.plugin.methods.inline_use_data import IInlineUseData
from metadata.scripts.paimon_moe import GACHA_LOG_PAIMON_MOE_PATH, update_paimon_moe_zh from metadata.scripts.paimon_moe import GACHA_LOG_PAIMON_MOE_PATH, update_paimon_moe_zh
from modules.gacha_log.const import UIGF_VERSION, GACHA_TYPE_LIST_REVERSE from modules.gacha_log.const import UIGF_VERSION, GACHA_TYPE_LIST_REVERSE
from modules.gacha_log.error import ( from modules.gacha_log.error import (
@ -589,3 +591,44 @@ class WishLogPlugin(Plugin.Conversation):
old_user_id: int, new_user_id: int, old_players: List["Player"] old_user_id: int, new_user_id: int, old_players: List["Player"]
) -> Optional[GachaLogMigrate]: ) -> Optional[GachaLogMigrate]:
return await GachaLogMigrate.create(old_user_id, new_user_id, old_players) return await GachaLogMigrate.create(old_user_id, new_user_id, old_players)
async def wish_log_use_by_inline(
self, update: "Update", context: "ContextTypes.DEFAULT_TYPE", pool_type: "BannerType"
):
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, "抽卡记录命令请求 || 参数 %s", pool_type.name if pool_type else None)
notice = None
try:
render_result = await self.rander_wish_log_analysis(user_id, uid, pool_type)
if isinstance(render_result, str):
notice = render_result
else:
await render_result.edit_inline_media(callback_query, filename="抽卡统计.png")
except GachaLogNotFound:
self.log_user(update, logger.info, "未找到抽卡记录")
notice = "未找到抽卡记录"
if notice:
await callback_query.answer(notice, show_alert=True)
async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]:
types = {
"角色": BannerType.CHARACTER1,
"武器": BannerType.WEAPON,
"常驻": BannerType.STANDARD,
"集录": BannerType.CHRONICLED,
}
data = []
for k, v in types.items():
data.append(
IInlineUseData(
text=f"{k}祈愿",
hash=f"wish_log_{v.value}",
callback=partial(self.wish_log_use_by_inline, pool_type=v),
player=True,
)
)
return data

View File

@ -112,7 +112,7 @@ class ErrorHandler(Plugin):
else: else:
buttons = ReplyKeyboardRemove() buttons = ReplyKeyboardRemove()
if chat.id == user.id: if (not chat) or chat.id == user.id:
logger.info("用户 %s[%s] 尝试通知错误信息[%s]", user.full_name, user.id, content) logger.info("用户 %s[%s] 尝试通知错误信息[%s]", user.full_name, user.id, content)
else: else:
self.log_user(update, logger.info, "尝试通知在 %s[%s] 的错误信息[%s]", chat.title, chat.id, content) self.log_user(update, logger.info, "尝试通知在 %s[%s] 的错误信息[%s]", chat.title, chat.id, content)

14
utils/patch/telegram.py Normal file
View File

@ -0,0 +1,14 @@
import telegram
from utils.patch.methods import patch, patchable
# https://github.com/python-telegram-bot/python-telegram-bot/issues/4295
@patch(telegram.Bot)
class Bot:
@patchable
def _effective_inline_results(self, results, next_offset=None, current_offset=None):
if current_offset == "[]":
current_offset = 50
return self.old__effective_inline_results(results, next_offset, current_offset)