✨ Support migrate user data

.devcontainer/devcontainer.json

@@ -0,0 +1,26 @@
+// For format details, see https://aka.ms/devcontainer.json. For config options, see the
+// README at: https://github.com/devcontainers/templates/tree/main/src/python
+{
+	"name": "Python 3",
+	// Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
+	"image": "mcr.microsoft.com/devcontainers/python:1-3.12-bookworm",
+	"features": {
+		"ghcr.io/devcontainers-contrib/features/pdm:2": {},
+		"ghcr.io/devcontainers-contrib/features/poetry:2": {}
+	},
+
+	// Features to add to the dev container. More info: https://containers.dev/features.
+	// "features": {},
+
+	// Use 'forwardPorts' to make a list of ports inside the container available locally.
+	// "forwardPorts": [],
+
+	// Use 'postCreateCommand' to run commands after the container is created.
+	"postCreateCommand": "poetry install --extras all"
+
+	// Configure tool-specific properties.
+	// "customizations": {},
+
+	// Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root.
+	// "remoteUser": "root"
+}

core/services/players/services.py

@@ -151,3 +151,9 @@ class PlayerInfoService(BaseService):
 
     async def delete(self, player_info: PlayerInfoSQLModel):
         await self._players_info_repository.delete(player_info)
+
+    async def update(self, player_info: PlayerInfoSQLModel):
+        await self._players_info_repository.update(player_info)
+
+    async def get_all_by_user_id(self, user_id: int):
+        return await self._players_info_repository.get_all_by_user_id(user_id)

gram_core

@@ -1 +1 @@
-Subproject commit 9dd1acab7842fb2c7cfd5ca507de93778318dc82
+Subproject commit 4718860a87e5b64eb59966f7122716fd70ece8f0

modules/gacha_log/log.py

@@ -107,6 +107,23 @@ class GachaLog:
             return True
         return False
 
+    async def move_history_info(self, user_id: str, uid: str, new_user_id: str) -> bool:
+        """移动历史抽卡记录数据
+        :param user_id: 用户id
+        :param uid: 原神uid
+        :param new_user_id: 新用户id
+        :return: 是否移动成功
+        """
+        old_file_path = self.gacha_log_path / f"{user_id}-{uid}.json"
+        new_file_path = self.gacha_log_path / f"{new_user_id}-{uid}.json"
+        if (not old_file_path.exists()) or new_file_path.exists():
+            return False
+        try:
+            old_file_path.rename(new_file_path)
+            return True
+        except PermissionError:
+            return False
+
     async def save_gacha_log_info(self, user_id: str, uid: str, info: GachaLogInfo):
         """保存抽卡记录数据
         :param user_id: 用户id

modules/gacha_log/migrate.py

@@ -0,0 +1,50 @@
+from typing import Optional, List, TYPE_CHECKING
+
+from gram_core.plugin.methods.migrate_data import IMigrateData, MigrateDataException
+from modules.gacha_log.log import GachaLog
+
+if TYPE_CHECKING:
+    from gram_core.services.players.models import Player
+
+
+class GachaLogMigrate(IMigrateData, GachaLog):
+    old_user_id: int
+    new_user_id: int
+    old_uid_list: List[int]
+
+    async def migrate_data_msg(self) -> str:
+        return f"{len(self.old_uid_list)} 个账号的抽卡记录数据"
+
+    async def migrate_data(self) -> bool:
+        uid_list = []
+        for uid in self.old_uid_list:
+            if not await self.move_history_info(str(self.old_user_id), str(uid), str(self.new_user_id)):
+                uid_list.append(str(uid))
+        if uid_list:
+            raise MigrateDataException(f"抽卡记录数据迁移失败：uid {','.join(uid_list)}")
+        return True
+
+    @classmethod
+    async def create(
+        cls,
+        old_user_id: int,
+        new_user_id: int,
+        players: List["Player"],
+    ) -> Optional["GachaLogMigrate"]:
+        if not players:
+            return None
+        _uid_list = [player.player_id for player in players if player and player.player_id]
+        if not _uid_list:
+            return None
+        self = cls()
+        old_uid_list = []
+        for uid in _uid_list:
+            _, status = await self.load_history_info(str(old_user_id), str(uid), True)
+            if status:
+                old_uid_list.append(uid)
+        if not old_uid_list:
+            return None
+        self.old_user_id = old_user_id
+        self.new_user_id = new_user_id
+        self.old_uid_list = old_uid_list
+        return self

modules/pay_log/log.py

@@ -96,6 +96,23 @@ class PayLog:
             return True
         return False
 
+    async def move_history_info(self, user_id: str, uid: str, new_user_id: str) -> bool:
+        """移动历史充值记录数据
+        :param user_id: 用户id
+        :param uid: 原神uid
+        :param new_user_id: 新用户id
+        :return: 是否移动成功
+        """
+        old_file_path = self.get_file_path(user_id, uid)
+        new_file_path = self.get_file_path(new_user_id, uid)
+        if (not old_file_path.exists()) or new_file_path.exists():
+            return False
+        try:
+            old_file_path.rename(new_file_path)
+            return True
+        except PermissionError:
+            return False
+
     async def save_pay_log_info(self, user_id: str, uid: str, info: PayLogModel) -> None:
         """保存日志记录数据
         :param user_id: 用户id

modules/pay_log/migrate.py

@@ -0,0 +1,50 @@
+from typing import Optional, List, TYPE_CHECKING
+
+from gram_core.plugin.methods.migrate_data import IMigrateData, MigrateDataException
+from modules.pay_log.log import PayLog
+
+if TYPE_CHECKING:
+    from gram_core.services.players.models import Player
+
+
+class PayLogMigrate(IMigrateData, PayLog):
+    old_user_id: int
+    new_user_id: int
+    old_uid_list: List[int]
+
+    async def migrate_data_msg(self) -> str:
+        return f"{len(self.old_uid_list)} 个账号的充值记录数据"
+
+    async def migrate_data(self) -> bool:
+        uid_list = []
+        for uid in self.old_uid_list:
+            if not await self.move_history_info(str(self.old_user_id), str(uid), str(self.new_user_id)):
+                uid_list.append(str(uid))
+        if uid_list:
+            raise MigrateDataException(f"充值记录数据迁移失败：uid {','.join(uid_list)}")
+        return True
+
+    @classmethod
+    async def create(
+        cls,
+        old_user_id: int,
+        new_user_id: int,
+        players: List["Player"],
+    ) -> Optional["PayLogMigrate"]:
+        if not players:
+            return None
+        _uid_list = [player.player_id for player in players if player and player.player_id]
+        if not _uid_list:
+            return None
+        self = cls()
+        old_uid_list = []
+        for uid in _uid_list:
+            _, status = await self.load_history_info(str(old_user_id), str(uid), True)
+            if status:
+                old_uid_list.append(uid)
+        if not old_uid_list:
+            return None
+        self.old_user_id = old_user_id
+        self.new_user_id = new_user_id
+        self.old_uid_list = old_uid_list
+        return self

plugins/account/account.py

@@ -19,6 +19,7 @@ from core.services.cookies.error import TooManyRequestPublicCookies, CookiesCach
 from core.services.cookies.services import CookiesService, PublicCookiesService
 from core.services.players.models import PlayersDataBase as Player, PlayerInfoSQLModel
 from core.services.players.services import PlayersService, PlayerInfoService
+from plugins.account.migrate import AccountMigrate
 from plugins.tools.genshin import GenshinHelper
 from utils.log import logger
 
@@ -292,3 +293,8 @@ class BindAccountPlugin(Plugin.Conversation):
             return ConversationHandler.END
         await message.reply_text("回复错误，请重新输入")
         return COMMAND_RESULT
+
+    async def get_migrate_data(self, old_user_id: int, new_user_id: int, _) -> Optional[AccountMigrate]:
+        return await AccountMigrate.create(
+            old_user_id, new_user_id, self.players_service, self.player_info_service, self.cookies_service
+        )

plugins/account/migrate.py

@@ -0,0 +1,137 @@
+from typing import Optional, List
+
+from sqlalchemy.orm.exc import StaleDataError
+
+from core.services.players.services import PlayerInfoService
+from gram_core.plugin.methods.migrate_data import IMigrateData, MigrateDataException
+from gram_core.services.cookies import CookiesService
+from gram_core.services.cookies.models import CookiesDataBase as Cookies
+from gram_core.services.players import PlayersService
+from gram_core.services.players.models import PlayersDataBase as Player, PlayerInfoSQLModel as PlayerInfo
+
+
+class AccountMigrate(IMigrateData):
+    old_user_id: int
+    new_user_id: int
+    players_service: PlayersService
+    player_info_service: PlayerInfoService
+    cookies_service: CookiesService
+    need_migrate_player: List[Player]
+    need_migrate_player_info: List[PlayerInfo]
+    need_migrate_cookies: List[Cookies]
+
+    async def migrate_data_msg(self) -> str:
+        text = []
+        if self.need_migrate_player:
+            text.append(f"player 数据 {len(self.need_migrate_player)} 条")
+        if self.need_migrate_player_info:
+            text.append(f"player_info 数据 {len(self.need_migrate_player_info)} 条")
+        if self.need_migrate_cookies:
+            text.append(f"cookies 数据 {len(self.need_migrate_cookies)} 条")
+        return "、".join(text)
+
+    async def migrate_data(self) -> bool:
+        players, players_info, cookies = [], [], []
+        for player in self.need_migrate_player:
+            try:
+                await self.players_service.update(player)
+            except StaleDataError:
+                players.append(str(player.player_id))
+        for player_info in self.need_migrate_player_info:
+            try:
+                await self.player_info_service.update(player_info)
+            except StaleDataError:
+                players_info.append(str(player_info.player_id))
+        for cookie in self.need_migrate_cookies:
+            try:
+                await self.cookies_service.update(cookie)
+            except StaleDataError:
+                cookies.append(str(cookie.account_id))
+        if any([players, players_info, cookies]):
+            text = []
+            if players:
+                text.append(f"player 数据迁移失败 player_id {','.join(players)}")
+            if players_info:
+                text.append(f"player_info 数据迁移失败 player_id {','.join(players_info)}")
+            if cookies:
+                text.append(f"cookies 数据迁移失败 account_id {','.join(cookies)}")
+            raise MigrateDataException("、".join(text))
+        return True
+
+    @staticmethod
+    async def create_players(
+        old_user_id: int,
+        new_user_id: int,
+        players_service: PlayersService,
+    ) -> List[Player]:
+        need_migrate, new_data = await AccountMigrate.filter_sql_data(
+            Player,
+            players_service.get_all_by_user_id,
+            old_user_id,
+            new_user_id,
+            (Player.account_id, Player.player_id, Player.region),
+        )
+        for i in need_migrate:
+            i.user_id = new_user_id
+            if new_data:
+                i.is_chosen = False
+        return need_migrate
+
+    @staticmethod
+    async def create_players_info(
+        old_user_id: int,
+        new_user_id: int,
+        player_info_service: PlayerInfoService,
+    ) -> List[PlayerInfo]:
+        need_migrate, _ = await AccountMigrate.filter_sql_data(
+            PlayerInfo,
+            player_info_service.get_all_by_user_id,
+            old_user_id,
+            new_user_id,
+            (PlayerInfo.user_id, PlayerInfo.player_id),
+        )
+        for i in need_migrate:
+            i.user_id = new_user_id
+        return need_migrate
+
+    @staticmethod
+    async def create_cookies(
+        old_user_id: int,
+        new_user_id: int,
+        cookies_service: CookiesService,
+    ) -> List[Cookies]:
+        need_migrate, _ = await AccountMigrate.filter_sql_data(
+            Cookies,
+            cookies_service.get_all,
+            old_user_id,
+            new_user_id,
+            (Cookies.user_id, Cookies.account_id, Cookies.region),
+        )
+        for i in need_migrate:
+            i.user_id = new_user_id
+        return need_migrate
+
+    @classmethod
+    async def create(
+        cls,
+        old_user_id: int,
+        new_user_id: int,
+        players_service: PlayersService,
+        player_info_service: PlayerInfoService,
+        cookies_service: CookiesService,
+    ) -> Optional["AccountMigrate"]:
+        need_migrate_player = await cls.create_players(old_user_id, new_user_id, players_service)
+        need_migrate_player_info = await cls.create_players_info(old_user_id, new_user_id, player_info_service)
+        need_migrate_cookies = await cls.create_cookies(old_user_id, new_user_id, cookies_service)
+        if not any([need_migrate_player, need_migrate_player_info, need_migrate_cookies]):
+            return None
+        self = cls()
+        self.old_user_id = old_user_id
+        self.new_user_id = new_user_id
+        self.players_service = players_service
+        self.player_info_service = player_info_service
+        self.cookies_service = cookies_service
+        self.need_migrate_player = need_migrate_player
+        self.need_migrate_player_info = need_migrate_player_info
+        self.need_migrate_cookies = need_migrate_cookies
+        return self

plugins/admin/migrate.py

@@ -0,0 +1,122 @@
+from functools import partial
+from typing import Dict, List, TYPE_CHECKING
+
+from telegram import InlineKeyboardButton, InlineKeyboardMarkup
+
+from core.plugin import Plugin, handler
+from gram_core.plugin.methods.migrate_data import IMigrateData, MigrateDataException
+from gram_core.services.players import PlayersService
+from utils.log import logger
+
+if TYPE_CHECKING:
+    from telegram import Update
+    from telegram.ext import ContextTypes
+
+
+class MigrateAdmin(Plugin):
+    def __init__(self, players_service: PlayersService):
+        self.players_service = players_service
+        self.cache_data: Dict[int, List[IMigrateData]] = {}
+        self.wait_time = 60
+
+    async def _add_pop_cache_job(self, user_id: int) -> None:
+        if user_id in self.cache_data:
+            del self.cache_data[user_id]
+
+    def add_pop_cache_job(self, user_id: int) -> None:
+        job_queue = self.application.job_queue
+        if job_queue is None:
+            raise RuntimeError
+        job_queue.run_once(
+            callback=partial(self._add_pop_cache_job, user_id=user_id), when=60, name=f"{user_id}|migrate_pop_cache"
+        )
+
+    def cancel_pop_cache_job(self, user_id: int) -> None:
+        job_queue = self.application.job_queue
+        if job_queue is None:
+            raise RuntimeError
+        if job := job_queue.get_jobs_by_name(f"{user_id}|migrate_pop_cache"):
+            job[0].schedule_removal()
+
+    @handler.command(command="migrate_admin", block=False, admin=True)
+    async def migrate_admin_command(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"):
+        message = update.effective_message
+        args = self.get_args(context)
+        logger.info("管理员 %s[%s] migrate_admin 命令请求", message.from_user.full_name, message.from_user.id)
+        if (not args) or len(args) < 2:
+            await message.reply_text("参数错误，请指定新旧用户 id ！")
+            return
+        try:
+            old_user_id, new_user_id = int(args[0]), int(args[1])
+        except ValueError:
+            await message.reply_text("参数错误，请指定新旧用户 id ！")
+            return
+        if old_user_id in self.cache_data:
+            await message.reply_text("该用户正在迁移数据中，请稍后再试！")
+            return
+        data = []
+        players = await self.players_service.get_all_by_user_id(old_user_id)
+        for _, instance in self.application.managers.plugins_map.items():
+            if _data := await instance.get_migrate_data(old_user_id, new_user_id, players):
+                data.append(_data)
+        if not data:
+            await message.reply_text("没有需要迁移的数据！")
+            return
+        text = "确定迁移以下数据？\n\n"
+        for d in data:
+            text += f"- {await d.migrate_data_msg()}\n"
+        self.cache_data[old_user_id] = data
+        buttons = [
+            [
+                InlineKeyboardButton(
+                    "确定迁移",
+                    callback_data=f"migrate_admin|{old_user_id}",
+                )
+            ],
+        ]
+        await message.reply_text(text, reply_markup=InlineKeyboardMarkup(buttons))
+        self.add_pop_cache_job(old_user_id)
+
+    async def try_migrate_data(self, user_id: int) -> str:
+        text = []
+        for d in self.cache_data[user_id]:
+            try:
+                logger.info("开始迁移数据 class[%s]", d.__class__.__name__)
+                await d.migrate_data()
+                logger.info("迁移数据成功 class[%s]", d.__class__.__name__)
+            except MigrateDataException as e:
+                text.append(e.msg)
+            except Exception as e:
+                logger.exception("迁移数据失败，未知错误！ class[%s]", d.__class__.__name__, exc_info=e)
+                text.append("迁移部分数据出现未知错误，请联系管理员！")
+        if text:
+            return "- " + "\n- ".join(text)
+
+    @handler.callback_query(pattern=r"^migrate_admin\|", block=False)
+    async def callback_query_migrate_admin(self, update: "Update", _: "ContextTypes.DEFAULT_TYPE") -> None:
+        callback_query = update.callback_query
+        user = callback_query.from_user
+        message = callback_query.message
+        logger.info("管理员 %s[%s] migrate_admin callback 请求", user.full_name, user.id)
+
+        async def get_migrate_admin_callback(callback_query_data: str) -> int:
+            _data = callback_query_data.split("|")
+            _old_user_id = int(_data[1])
+            logger.debug("callback_query_data函数返回 old_user_id[%s]", _old_user_id)
+            return _old_user_id
+
+        old_user_id = await get_migrate_admin_callback(callback_query.data)
+        if old_user_id not in self.cache_data:
+            await callback_query.answer("请求已过期，请重新发起请求！", show_alert=True)
+            self.add_delete_message_job(message, delay=5)
+            return
+        self.cancel_pop_cache_job(old_user_id)
+        await message.edit_text("正在迁移数据，请稍后...", reply_markup=None)
+        try:
+            text = await self.try_migrate_data(old_user_id)
+        finally:
+            await self._add_pop_cache_job(old_user_id)
+        if text:
+            await message.edit_text(f"迁移部分数据失败！\n\n{text}")
+            return
+        await message.edit_text("迁移数据成功！")

plugins/genshin/pay_log.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional, List
 
 from simnet import GenshinClient, Region
 from telegram import InlineKeyboardButton, InlineKeyboardMarkup
@@ -14,6 +14,7 @@ from core.services.template.services import TemplateService
 from modules.gacha_log.helpers import from_url_get_authkey
 from modules.pay_log.error import PayLogNotFound, PayLogAccountNotFound, PayLogInvalidAuthkey, PayLogAuthkeyTimeout
 from modules.pay_log.log import PayLog
+from modules.pay_log.migrate import PayLogMigrate
 from plugins.tools.genshin import PlayerNotFoundError, CookiesNotFoundError
 from plugins.tools.player_info import PlayerInfoSystem
 from utils.log import logger
@@ -21,6 +22,7 @@ from utils.log import logger
 if TYPE_CHECKING:
     from telegram import Update, User
     from telegram.ext import ContextTypes
+    from gram_core.services.players.models import Player
 
 
 INPUT_URL, CONFIRM_DELETE = range(10100, 10102)
@@ -96,9 +98,9 @@ class PayLogPlugin(Plugin.Conversation):
                     await message.reply_text("该功能需要绑定 stoken 才能使用")
                     return ConversationHandler.END
             else:
-                raise CookiesNotFoundError
+                raise CookiesNotFoundError(user.id)
         else:
-            raise CookiesNotFoundError
+            raise CookiesNotFoundError(user.id)
         text = "小派蒙正在从服务器获取数据，请稍后"
         reply = await message.reply_text(text)
         await message.reply_chat_action(ChatAction.TYPING)
@@ -229,3 +231,9 @@ class PayLogPlugin(Plugin.Conversation):
                 [InlineKeyboardButton("点我导入", url=create_deep_linked_url(context.bot.username, "pay_log_import"))]
             ]
             await message.reply_text("派蒙没有找到你的充值记录，快来点击按钮私聊派蒙导入吧~", reply_markup=InlineKeyboardMarkup(buttons))
+
+    @staticmethod
+    async def get_migrate_data(
+        old_user_id: int, new_user_id: int, old_players: List["Player"]
+    ) -> Optional[PayLogMigrate]:
+        return await PayLogMigrate.create(old_user_id, new_user_id, old_players)

plugins/genshin/wish_log.py

@@ -1,5 +1,5 @@
 from io import BytesIO
-from typing import Optional, TYPE_CHECKING
+from typing import Optional, TYPE_CHECKING, List
 from urllib.parse import urlencode
 
 from aiofiles import open as async_open
@@ -29,6 +29,7 @@ from modules.gacha_log.error import (
 )
 from modules.gacha_log.helpers import from_url_get_authkey
 from modules.gacha_log.log import GachaLog
+from modules.gacha_log.migrate import GachaLogMigrate
 from plugins.tools.genshin import PlayerNotFoundError
 from plugins.tools.player_info import PlayerInfoSystem
 from utils.log import logger
@@ -43,6 +44,7 @@ except ImportError:
 if TYPE_CHECKING:
     from telegram import Update, Message, User, Document
     from telegram.ext import ContextTypes
+    from gram_core.services.players.models import Player
 
 INPUT_URL, INPUT_FILE, CONFIRM_DELETE = range(10100, 10103)
 
@@ -447,3 +449,9 @@ class WishLogPlugin(Plugin.Conversation):
                 [InlineKeyboardButton("点我导入", url=create_deep_linked_url(context.bot.username, "gacha_log_import"))]
             ]
             await message.reply_text("派蒙没有找到你的抽卡记录，快来私聊派蒙导入吧~", reply_markup=InlineKeyboardMarkup(buttons))
+
+    @staticmethod
+    async def get_migrate_data(
+        old_user_id: int, new_user_id: int, old_players: List["Player"]
+    ) -> Optional[GachaLogMigrate]:
+        return await GachaLogMigrate.create(old_user_id, new_user_id, old_players)

plugins/tools/daily_note.py

@@ -11,12 +11,14 @@ from telegram.error import BadRequest, Forbidden
 from core.plugin import Plugin
 from core.services.task.models import Task as TaskUser, TaskStatusEnum
 from core.services.task.services import TaskResinServices, TaskRealmServices, TaskExpeditionServices
+from gram_core.plugin.methods.migrate_data import IMigrateData, MigrateDataException
 from plugins.tools.genshin import GenshinHelper, PlayerNotFoundError, CookiesNotFoundError
 from utils.log import logger
 
 if TYPE_CHECKING:
     from simnet import GenshinClient
     from telegram.ext import ContextTypes
+    from gram_core.services.task.services import TaskServices
 
 
 class TaskDataBase(BaseModel):
@@ -368,3 +370,65 @@ class DailyNoteSystem(Plugin):
                 else:
                     task_user_db.status = TaskStatusEnum.STATUS_SUCCESS
             await self.update_task_user(task_db)
+
+    async def get_migrate_data(self, old_user_id: int, new_user_id: int, _) -> Optional["TaskMigrate"]:
+        return await TaskMigrate.create(
+            old_user_id,
+            new_user_id,
+            self.resin_service,
+        )
+
+
+class TaskMigrate(IMigrateData):
+    old_user_id: int
+    new_user_id: int
+    task_services: "TaskServices"
+    need_migrate_tasks: List[TaskUser]
+
+    async def migrate_data_msg(self) -> str:
+        return f"定时任务数据 {len(self.need_migrate_tasks)} 条"
+
+    async def migrate_data(self) -> bool:
+        id_list = []
+        for task in self.need_migrate_tasks:
+            try:
+                await self.task_services.update(task)
+            except StaleDataError:
+                id_list.append(str(task.id))
+        if id_list:
+            raise MigrateDataException(f"任务数据迁移失败：id {','.join(id_list)}")
+        return True
+
+    @staticmethod
+    async def create_tasks(
+        old_user_id: int,
+        new_user_id: int,
+        task_services: "TaskServices",
+    ) -> List[TaskUser]:
+        need_migrate, _ = await TaskMigrate.filter_sql_data(
+            TaskUser,
+            task_services.get_all_by_user_id,
+            old_user_id,
+            new_user_id,
+            (TaskUser.user_id, TaskUser.type),
+        )
+        for i in need_migrate:
+            i.user_id = new_user_id
+        return need_migrate
+
+    @classmethod
+    async def create(
+        cls,
+        old_user_id: int,
+        new_user_id: int,
+        task_services: "TaskServices",
+    ) -> Optional["TaskMigrate"]:
+        need_migrate_tasks = await cls.create_tasks(old_user_id, new_user_id, task_services)
+        if not need_migrate_tasks:
+            return None
+        self = cls()
+        self.old_user_id = old_user_id
+        self.new_user_id = new_user_id
+        self.task_services = task_services
+        self.need_migrate_tasks = need_migrate_tasks
+        return self