PaiGram/plugins/genshin/calendar.py

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

91 lines
3.5 KiB
Python
Raw Permalink Normal View History

2023-02-16 09:38:43 +00:00
from datetime import datetime, timedelta
2024-06-19 08:29:47 +00:00
from functools import partial
from typing import Dict, List, Optional, TYPE_CHECKING
2023-02-16 08:47:48 +00:00
from telegram import Update
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, MessageHandler, filters
from core.dependence.assets import AssetsService
from core.dependence.redisdb import RedisDB
2023-02-16 08:47:48 +00:00
from core.plugin import Plugin, handler
from core.services.template.services import TemplateService
2024-06-19 08:29:47 +00:00
from gram_core.plugin.methods.inline_use_data import IInlineUseData
2023-02-16 08:47:48 +00:00
from modules.apihelper.client.components.calendar import Calendar
from utils.log import logger
try:
import ujson as jsonlib
except ImportError:
import json as jsonlib
2024-06-19 08:29:47 +00:00
if TYPE_CHECKING:
from telegram.ext import ContextTypes
from gram_core.services.template.models import RenderResult
2023-02-16 08:47:48 +00:00
class CalendarPlugin(Plugin):
2023-02-16 08:47:48 +00:00
"""活动日历查询"""
def __init__(
self,
template_service: TemplateService,
assets_service: AssetsService,
redis: RedisDB,
2023-02-16 08:47:48 +00:00
):
self.template_service = template_service
self.assets_service = assets_service
self.calendar = Calendar()
self.cache = redis.client
async def _fetch_data(self) -> Dict:
if data := await self.cache.get("plugin:calendar"):
return jsonlib.loads(data.decode("utf-8"))
data = await self.calendar.get_photo_data(self.assets_service)
2023-02-16 09:38:43 +00:00
now = datetime.now()
next_hour = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
await self.cache.set("plugin:calendar", jsonlib.dumps(data, default=lambda x: x.dict()), ex=next_hour - now)
2023-02-16 08:47:48 +00:00
return data
2024-06-19 08:29:47 +00:00
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",
)
2023-02-16 08:47:48 +00:00
@handler.command("calendar", block=False)
@handler(MessageHandler, filters=filters.Regex(r"^(活动)+(日历|日历列表)$"), block=False)
async def command_start(self, update: Update, _: CallbackContext) -> None:
message = update.effective_message
mode = "list" if "列表" in message.text else "calendar"
2024-03-10 12:40:26 +00:00
self.log_user(update, logger.info, "查询日历 | 模式 %s", mode)
2023-02-16 08:47:48 +00:00
await message.reply_chat_action(ChatAction.TYPING)
2024-06-19 08:29:47 +00:00
image = await self.render(mode == "list")
2023-02-16 08:47:48 +00:00
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
await image.reply_photo(message)
2024-06-19 08:29:47 +00:00
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),
),
]