[Enhance] Self account delete

- Add support for users destroy their own accounts

latest version/core/config_manager.py

@@ -71,6 +71,8 @@ class Config:
 
     SAVE_FULL_UNLOCK = False
 
+    ALLOW_SELF_ACCOUNT_DELETE = False
+
     # ------------------------------------------
 
     # You can change this to make another PTT mechanism.
@@ -86,6 +88,7 @@ class Config:
     SQLITE_DATABASE_BACKUP_FOLDER_PATH = './database/backup/'
     DATABASE_INIT_PATH = './database/init/'
     SQLITE_LOG_DATABASE_PATH = './database/arcaea_log.db'
+    SQLITE_DATABASE_DELETED_PATH = './database/arcaea_database_deleted.db'
 
     GAME_LOGIN_RATE_LIMIT = '30/5 minutes'
     API_LOGIN_RATE_LIMIT = '10/5 minutes'

latest version/core/constant.py

@@ -1,6 +1,6 @@
 from .config_manager import Config
 
-ARCAEA_SERVER_VERSION = 'v2.11.3.3'
+ARCAEA_SERVER_VERSION = 'v2.11.3.4'
 ARCAEA_LOG_DATBASE_VERSION = 'v1.1'
 
 
@@ -45,6 +45,7 @@ class Constant:
     SONGLIST_FILE_PATH = Config.SONGLIST_FILE_PATH
     SQLITE_DATABASE_PATH = Config.SQLITE_DATABASE_PATH
     SQLITE_LOG_DATABASE_PATH = Config.SQLITE_LOG_DATABASE_PATH
+    SQLITE_DATABASE_DELETED_PATH = Config.SQLITE_DATABASE_DELETED_PATH
 
     DOWNLOAD_TIMES_LIMIT = Config.DOWNLOAD_TIMES_LIMIT
     DOWNLOAD_TIME_GAP_LIMIT = Config.DOWNLOAD_TIME_GAP_LIMIT

latest version/core/init.py

@@ -177,6 +177,16 @@ class LogDatabaseInit:
             self.table_init()
 
 
+class DeletedDatabaseInit(DatabaseInit):
+    def __init__(self, db_path: str = Config.SQLITE_DATABASE_DELETED_PATH) -> None:
+        super().__init__(db_path)
+
+    def init(self) -> None:
+        with Connect(self.db_path) as c:
+            self.c = c
+            self.table_init()
+
+
 class FileChecker:
     '''文件检查及初始化类'''
 
@@ -195,7 +205,7 @@ class FileChecker:
             self.logger.warning('Folder `%s` is missing.' % folder_path)
         return f
 
-    def check_update_database(self) -> bool:
+    def _check_update_database_log(self) -> bool:
         if not self.check_file(Config.SQLITE_LOG_DATABASE_PATH):
             # 新建日志数据库
             try:
@@ -232,22 +242,22 @@ class FileChecker:
                         f'Failed to update the file `{Config.SQLITE_LOG_DATABASE_PATH}`')
                     return False
 
-        if not self.check_file(Config.SQLITE_DATABASE_PATH):
+        return True
+
+    def _check_update_database_main(self, db_path=Config.SQLITE_DATABASE_PATH, init_class=DatabaseInit) -> bool:
+        if not self.check_file(db_path):
             # 新建数据库
             try:
-                self.logger.info(
-                    'Try to new the file `%s`.' % Config.SQLITE_DATABASE_PATH)
-                DatabaseInit().init()
-                self.logger.info(
-                    'Success to new the file `%s`.' % Config.SQLITE_DATABASE_PATH)
+                self.logger.info(f'Try to new the file `{db_path}`.')
+                init_class().init()
+                self.logger.info(f'Success to new the file `{db_path}`.')
             except Exception as e:
                 self.logger.error(format_exc())
-                self.logger.warning(
-                    'Failed to new the file `%s`.' % Config.SQLITE_DATABASE_PATH)
+                self.logger.warning(f'Failed to new the file `{db_path}`.')
                 return False
         else:
             # 检查更新
-            with Connect() as c:
+            with Connect(db_path) as c:
                 try:
                     c.execute('''select value from config where id="version"''')
                     x = c.fetchone()
@@ -256,42 +266,45 @@ class FileChecker:
             # 数据库自动更新，不强求
             if not x or x[0] != ARCAEA_SERVER_VERSION:
                 self.logger.warning(
-                    'Maybe the file `%s` is an old version.' % Config.SQLITE_DATABASE_PATH)
+                    f'Maybe the file `{db_path}` is an old version. Version: {x[0] if x else "None"}')
                 try:
                     self.logger.info(
-                        'Try to update the file `%s`.' % Config.SQLITE_DATABASE_PATH)
+                        f'Try to update the file `{db_path}` to version {ARCAEA_SERVER_VERSION}.')
 
                     if not os.path.isdir(Config.SQLITE_DATABASE_BACKUP_FOLDER_PATH):
                         os.makedirs(Config.SQLITE_DATABASE_BACKUP_FOLDER_PATH)
 
-                    backup_path = try_rename(Config.SQLITE_DATABASE_PATH, os.path.join(
-                        Config.SQLITE_DATABASE_BACKUP_FOLDER_PATH, os.path.split(Config.SQLITE_DATABASE_PATH)[-1] + '.bak'))
+                    backup_path = try_rename(db_path, os.path.join(
+                        Config.SQLITE_DATABASE_BACKUP_FOLDER_PATH, os.path.split(db_path)[-1] + '.bak'))
 
                     try:
-                        copy2(backup_path, Config.SQLITE_DATABASE_PATH)
+                        copy2(backup_path, db_path)
                     except:
-                        copy(backup_path, Config.SQLITE_DATABASE_PATH)
+                        copy(backup_path, db_path)
 
                     temp_path = os.path.join(
-                        *os.path.split(Config.SQLITE_DATABASE_PATH)[:-1], 'old_arcaea_database.db')
+                        *os.path.split(db_path)[:-1], 'old_arcaea_database.db')
                     if os.path.isfile(temp_path):
                         os.remove(temp_path)
 
-                    try_rename(Config.SQLITE_DATABASE_PATH, temp_path)
+                    try_rename(db_path, temp_path)
 
-                    DatabaseInit().init()
-                    self.update_database(temp_path)
+                    init_class().init()
+                    self.update_database(temp_path, db_path)
 
                     self.logger.info(
-                        'Success to update the file `%s`.' % Config.SQLITE_DATABASE_PATH)
+                        f'Success to update the file `{db_path}`.')
 
                 except Exception as e:
                     self.logger.error(format_exc())
                     self.logger.warning(
-                        'Fail to update the file `%s`.' % Config.SQLITE_DATABASE_PATH)
+                        f'Fail to update the file `{db_path}`.')
 
         return True
 
+    def check_update_database(self) -> bool:
+        return self._check_update_database_main() and self._check_update_database_log() and self._check_update_database_main(Config.SQLITE_DATABASE_DELETED_PATH, DeletedDatabaseInit)
+
     @staticmethod
     def update_database(old_path: str, new_path: str = Config.SQLITE_DATABASE_PATH) -> None:
         '''更新数据库，并删除旧文件'''

latest version/core/operation.py

@@ -1,3 +1,4 @@
+from .constant import Constant
 from .download import DownloadList
 from .error import NoData
 from .save import SaveData
@@ -245,3 +246,75 @@ class UnlockUserItem(BaseOperation):
         with Connect() as c:
             c.execute(
                 f'''delete from user_item where type in ({','.join(['?'] * len(self.item_types))})''', self.item_types)
+
+
+def _delete_one_table(c, table_name, user_id):
+    c.execute(
+        f'''insert into db_deleted.{table_name} select * from {table_name} where user_id = ?''', (user_id,))
+    c.execute(f'''delete from {table_name} where user_id = ?''', (user_id,))
+
+
+class DeleteUserScore(BaseOperation):
+    '''
+        删除单用户成绩，不包含 recent 数据
+    '''
+    _name = 'delete_user_score'
+
+    def __init__(self, user=None):
+        self.user = user
+
+    def set_params(self, user_id: int = None, *args, **kwargs):
+        if user_id is not None:
+            self.user = User()
+            self.user.user_id = int(user_id)
+        return self
+
+    def run(self):
+        assert self.user is not None
+        with Connect() as c:
+            c.execute('''attach database ? as db_deleted''',
+                      (Constant.SQLITE_DATABASE_DELETED_PATH,))
+            _delete_one_table(c, 'best_score', self.user.user_id)
+            _delete_one_table(c, 'recent30', self.user.user_id)
+
+
+class DeleteOneUser(BaseOperation):
+    '''
+        删除单用户
+    '''
+    _name = 'delete_one_user'
+
+    TABLES = ['best_score', 'recent30', 'user_char', 'user_course', 'user_item',
+              'user_present', 'user_redeem', 'user_role', 'user_save', 'user_world', 'user']
+
+    def __init__(self, user=None):
+        self.user = user
+
+    def set_params(self, user_id: int = None, *args, **kwargs):
+        if user_id is not None:
+            self.user = User()
+            self.user.user_id = int(user_id)
+        return self
+
+    def run(self):
+        assert self.user is not None
+        with Connect() as c:
+            c.execute('''attach database ? as db_deleted''',
+                      (Constant.SQLITE_DATABASE_DELETED_PATH,))
+
+            self._clear_login(c)
+            self._data_save(c)
+
+    def _data_save(self, c):
+        c.execute(
+            f'''insert into db_deleted.friend select * from friend where user_id_me = ? or user_id_other = ?''', (self.user.user_id, self.user.user_id))
+        c.execute(f'''delete from friend where user_id_me = ? or user_id_other = ?''',
+                  (self.user.user_id, self.user.user_id))
+
+        [_delete_one_table(c, x, self.user.user_id) for x in self.TABLES]
+
+    def _clear_login(self, c):
+        c.execute('''delete from login where user_id = ?''',
+                  (self.user.user_id,))
+        c.execute('''delete from api_login where user_id = ?''',
+                  (self.user.user_id,))