diff --git a/plugins/admin/set_command.py b/plugins/admin/set_command.py index d351c91..8376767 100644 --- a/plugins/admin/set_command.py +++ b/plugins/admin/set_command.py @@ -29,6 +29,8 @@ class SetCommandPlugin(Plugin): BotCommand("player", "管理用户绑定玩家"), BotCommand("verify", "手动验证"), BotCommand("daily_note_tasks", "自动便笺提醒"), + BotCommand("cookies_import", "从其他 BOT 导入账号信息"), + BotCommand("cookies_export", "导出账号信息给其他 BOT"), ] group_command = [ BotCommand("help", "帮助"), @@ -55,8 +57,7 @@ class SetCommandPlugin(Plugin): BotCommand("rogue", "模拟宇宙信息查询"), BotCommand("rogue_locust", "寰宇蝗灾信息查询"), BotCommand("rogue_tourn", "差分宇宙信息查询"), - BotCommand("cookies_import", "从其他 BOT 导入账号信息"), - BotCommand("cookies_export", "导出账号信息给其他 BOT"), + BotCommand("activity_export", "活动数据导出"), ] admin_command = [ BotCommand("add_admin", "添加管理员"), diff --git a/plugins/starrail/activity.py b/plugins/starrail/activity.py new file mode 100644 index 0000000..036efad --- /dev/null +++ b/plugins/starrail/activity.py @@ -0,0 +1,90 @@ +import datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import aiofiles +import ujson +from simnet.models.starrail.chronicle.activity import StarRailActivity +from telegram.ext import filters + +from gram_core.plugin import Plugin, handler +from plugins.tools.genshin import GenshinHelper +from utils.log import logger +from pydantic import BaseModel + +if TYPE_CHECKING: + from telegram import Update + from telegram.ext import ContextTypes + + from simnet import StarRailClient + +data_path = Path("data") / "apihelper" / "activity" +data_path.mkdir(parents=True, exist_ok=True) + + +class ActivityUserModel(BaseModel): + uid: str = "0" + lang: str = "zh-cn" + region_time_zone: int = 8 + export_time: str = "" + export_timestamp: int = 0 + export_app: str = "PamGram" + export_app_version: str = "4.0" + + def __init__(self, **data: Any): + super().__init__(**data) + if not self.export_time: + self.export_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + self.export_timestamp = int(datetime.datetime.now().timestamp()) + + +class ActivityModel(StarRailActivity): + + info: ActivityUserModel + + +class NotHaveData(Exception): + """没有数据""" + + MSG = "没有查找到活动数据" + + +class PlayerActivityPlugins(Plugin): + """玩家活动信息查询""" + + def __init__( + self, + helper: GenshinHelper, + ): + self.helper = helper + + @handler.command("activity_export", block=False) + @handler.message(filters.Regex("^活动数据导出(.*)"), block=False) + async def activity_export(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE"): + user_id = await self.get_real_user_id(update) + message = update.effective_message + uid, offset = self.get_real_uid_or_offset(update) + self.log_user(update, logger.info, "导出活动数据命令请求") + try: + async with self.helper.genshin_or_public(user_id, uid=uid, offset=offset) as client: + client: "StarRailClient" + uid = client.player_id + data = await client.get_starrail_activity() + if not data.activities: + raise NotHaveData() + export_data = ActivityModel(info=ActivityUserModel(uid=str(uid)), **data.dict()) + except NotHaveData as e: + reply_message = await message.reply_text(e.MSG) + if filters.ChatType.GROUPS.filter(reply_message): + self.add_delete_message_job(message) + self.add_delete_message_job(reply_message) + return + filename = data_path / f"{uid}.json" + async with aiofiles.open(filename, "w", encoding="utf-8") as f: + await f.write(ujson.dumps(export_data.dict(), indent=4, ensure_ascii=False)) + + await message.reply_document( + document=filename, + filename=f"{uid}.json", + caption="活动数据导出成功", + )