PamGram/modules/pay_log/migrate.py
2023-12-16 18:10:42 +08:00

51 lines
1.6 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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