Arcaea-server/latest version/core/init.py

import os
import sys
from importlib import import_module
from json import load
from shutil import copy, copy2
from time import time
from traceback import format_exc

from core.bundle import BundleParser
from core.config_manager import Config
from core.constant import ARCAEA_DATABASE_VERSION, ARCAEA_LOG_DATBASE_VERSION
from core.course import Course
from core.download import DownloadList
from core.purchase import Purchase
from core.sql import (Connect, DatabaseMigrator, LogDatabaseMigrator,
                      MemoryDatabase)
from core.user import UserRegister
from core.util import try_rename
from core.world import MapParser


class DatabaseInit:
    def __init__(self, db_path: str = Config.SQLITE_DATABASE_PATH, init_folder_path: str = Config.DATABASE_INIT_PATH) -> None:
        self.db_path = db_path
        self.init_folder_path = init_folder_path
        self.c = None
        self.init_data = None

    @property
    def sql_path(self) -> str:
        return os.path.join(self.init_folder_path, 'tables.sql')

    @property
    def course_path(self) -> str:
        return os.path.join(self.init_folder_path, 'courses.json')

    @property
    def pack_path(self) -> str:
        return os.path.join(self.init_folder_path, 'packs.json')

    @property
    def single_path(self) -> str:
        return os.path.join(self.init_folder_path, 'singles.json')

    def table_init(self) -> None:
        '''初始化数据库结构'''
        with open(self.sql_path, 'r', encoding='utf-8') as f:
            self.c.executescript(f.read())
        self.c.execute('''insert into config values("version", :a);''', {
            'a': ARCAEA_DATABASE_VERSION})

    def character_init(self) -> None:
        '''初始化搭档信息'''
        uncapped_characters = self.init_data.char_core.keys()
        for i in range(0, len(self.init_data.char)):
            skill_requires_uncap = 1 if i == 2 else 0
            if i in uncapped_characters:
                max_level = 30
                uncapable = 1
            else:
                max_level = 20
                uncapable = 0
            sql = '''insert into character values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)'''
            self.c.execute(sql, (i, self.init_data.char[i], max_level, self.init_data.frag1[i], self.init_data.prog1[i], self.init_data.overdrive1[i], self.init_data.frag20[i], self.init_data.prog20[i], self.init_data.overdrive20[i],
                                 self.init_data.frag30[i], self.init_data.prog30[i], self.init_data.overdrive30[i], self.init_data.skill_id[i], self.init_data.skill_unlock_level[i], skill_requires_uncap, self.init_data.skill_id_uncap[i], self.init_data.char_type[i], uncapable))

        self.c.execute('''insert into character values(?,?,20,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0)''', (
            99, 'shirahime', 38, 33, 28, 66, 58, 50, 66, 58, 50, 'frags_preferred_song', 0, 0, '', 0))

        for i in self.init_data.char_core:
            self.c.executemany('''insert into char_item values(?,?,'core',?)''', [
                (i, j['core_id'], j['amount']) for j in self.init_data.char_core[i]])

    def insert_purchase_item(self, purchases: list):
        '''处理singles和packs'''
        for i in purchases:
            x = Purchase(self.c).from_dict(i)
            x.insert_all()

    def item_init(self) -> None:
        '''初始化物品信息'''
        self.c.executemany('''insert into item values(?,"core",1)''',
                           [(i,) for i in self.init_data.cores])
        self.c.executemany('''insert into item values(?,"world_song",1)''', [
            (i,) for i in self.init_data.world_songs])
        self.c.executemany('''insert into item values(?,"world_unlock",1)''', [
            (i,) for i in self.init_data.world_unlocks])
        self.c.executemany('''insert into item values(?,"course_banner",1)''', [
            (i,) for i in self.init_data.course_banners])

        self.c.execute('''insert into item values(?,?,?)''',
                       ('fragment', 'fragment', 1))
        self.c.execute('''insert into item values(?,?,?)''',
                       ('memory', 'memory', 1))
        self.c.execute('''insert into item values(?,?,?)''',
                       ('anni5tix', 'anni5tix', 1))
        self.c.execute('''insert into item values(?,?,?)''',
                       ('pick_ticket', 'pick_ticket', 1))

        with open(self.pack_path, 'rb') as f:
            self.insert_purchase_item(load(f))

        with open(self.single_path, 'rb') as f:
            self.insert_purchase_item(load(f))

        self.c.execute('''insert into item values(?,?,?)''',  # 新手任务奖励曲
                       ('innocence', 'world_song', 1))

    def course_init(self) -> None:
        '''初始化课题信息'''
        courses = []
        with open(self.course_path, 'rb') as f:
            courses = load(f)
        for i in courses:
            x = Course(self.c).from_dict(i)
            x.insert_all()

    def role_power_init(self) -> None:
        '''初始化power和role'''
        self.c.executemany('''insert into role values(?,?)''', [(
            self.init_data.role[i], self.init_data.role_caption[i]) for i in range(len(self.init_data.role))])

        self.c.executemany('''insert into power values(?,?)''', [(
            self.init_data.power[i], self.init_data.power_caption[i]) for i in range(len(self.init_data.power))])

        for i in self.init_data.role_power:
            self.c.executemany('''insert into role_power values(?,?)''', [
                               (i, j) for j in self.init_data.role_power[i]])

    def admin_init(self) -> None:
        '''初始化测试账号'''
        x = UserRegister(self.c)
        x.user_code = '123456789'
        x.user_id = 2000000
        x.name = 'admin'
        x.email = 'admin@admin.com'
        now = int(time() * 1000)

        x._insert_user_char()

        self.c.execute('''insert into user(user_id, name, password, join_date, user_code, rating_ptt,
        character_id, is_skill_sealed, is_char_uncapped, is_char_uncapped_override, is_hide_rating, favorite_character, max_stamina_notification_enabled, current_map, ticket, prog_boost, email)
        values(:user_id, :name, :password, :join_date, :user_code, 0, 0, 0, 0, 0, 0, -1, 0, '', :memories, 0, :email)
        ''', {'user_code': x.user_code, 'user_id': x.user_id, 'join_date': now, 'name': x.name, 'password': '41e5653fc7aeb894026d6bb7b2db7f65902b454945fa8fd65a6327047b5277fb', 'memories': 114514, 'email': x.email})

        self.c.execute(
            '''insert into user_role values(?, "admin")''', (x.user_id,))

    def init(self) -> None:
        sys.path.append(os.path.join(sys.path[0], self.init_folder_path))
        self.init_data = import_module('arc_data').InitData

        with Connect(self.db_path) as c:
            self.c = c
            self.table_init()
            self.character_init()
            self.item_init()
            self.course_init()
            self.role_power_init()
            self.admin_init()


class LogDatabaseInit:
    def __init__(self, db_path: str = Config.SQLITE_LOG_DATABASE_PATH, init_folder_path: str = Config.DATABASE_INIT_PATH) -> None:
        self.db_path = db_path
        self.init_folder_path = init_folder_path
        self.c = None

    @property
    def sql_path(self) -> str:
        return os.path.join(self.init_folder_path, 'log_tables.sql')

    def table_init(self) -> None:
        '''初始化数据库结构'''
        with open(self.sql_path, 'r') as f:
            self.c.executescript(f.read())
        self.c.execute(
            '''insert into cache values("version", :a, -1);''', {'a': ARCAEA_LOG_DATBASE_VERSION})

    def init(self) -> None:
        with Connect(self.db_path) as c:
            self.c = c
            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:
    '''文件检查及初始化类'''

    def __init__(self, logger=None):
        self.logger = logger

    def check_file(self, file_path: str) -> bool:
        f = os.path.isfile(file_path)
        if not f:
            self.logger.warning('File `%s` is missing.' % file_path)
        return f

    def check_folder(self, folder_path: str) -> bool:
        f = os.path.isdir(folder_path)
        if not f:
            self.logger.warning('Folder `%s` is missing.' % folder_path)
        return f

    def _check_update_database_log(self) -> bool:
        if not self.check_file(Config.SQLITE_LOG_DATABASE_PATH):
            # 新建日志数据库
            try:
                self.logger.info(
                    f'Try to new the file {Config.SQLITE_LOG_DATABASE_PATH}')
                LogDatabaseInit().init()
                self.logger.info(
                    f'Success to new the file {Config.SQLITE_LOG_DATABASE_PATH}')
            except Exception as e:
                self.logger.error(format_exc())
                self.logger.error(
                    f'Failed to new the file {Config.SQLITE_LOG_DATABASE_PATH}')
                return False
        else:
            # 检查更新
            with Connect(Config.SQLITE_LOG_DATABASE_PATH) as c:
                try:
                    x = c.execute(
                        '''select value from cache where key="version"''').fetchone()
                except:
                    x = None
            if not x or x[0] != ARCAEA_LOG_DATBASE_VERSION:
                self.logger.warning(
                    f'Maybe the file `{Config.SQLITE_LOG_DATABASE_PATH}` is an old version.')
                try:
                    self.logger.info(
                        f'Try to update the file `{Config.SQLITE_LOG_DATABASE_PATH}`')
                    self.update_log_database()
                    self.logger.info(
                        f'Success to update the file `{Config.SQLITE_LOG_DATABASE_PATH}`')
                except Exception as e:
                    self.logger.error(format_exc())
                    self.logger.error(
                        f'Failed to update the file `{Config.SQLITE_LOG_DATABASE_PATH}`')
                    return False

        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(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(f'Failed to new the file `{db_path}`.')
                return False
        else:
            # 检查更新
            with Connect(db_path) as c:
                try:
                    c.execute('''select value from config where id="version"''')
                    x = c.fetchone()
                except:
                    x = None
            # 数据库自动更新，不强求
            if not x or x[0] != ARCAEA_DATABASE_VERSION:
                self.logger.warning(
                    f'Maybe the file `{db_path}` is an old version. Version: {x[0] if x else "None"}')
                try:
                    self.logger.info(
                        f'Try to update the file `{db_path}` to version {ARCAEA_DATABASE_VERSION}.')

                    if not os.path.isdir(Config.SQLITE_DATABASE_BACKUP_FOLDER_PATH):
                        os.makedirs(Config.SQLITE_DATABASE_BACKUP_FOLDER_PATH)

                    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, db_path)
                    except:
                        copy(backup_path, db_path)

                    temp_path = os.path.join(
                        *os.path.split(db_path)[:-1], 'old_arcaea_database.db')
                    if os.path.isfile(temp_path):
                        os.remove(temp_path)

                    try_rename(db_path, temp_path)

                    init_class().init()
                    self.update_database(temp_path, db_path)

                    self.logger.info(
                        f'Success to update the file `{db_path}`.')

                except Exception as e:
                    self.logger.error(format_exc())
                    self.logger.warning(
                        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:
        '''更新数据库，并删除旧文件'''
        if os.path.isfile(old_path) and os.path.isfile(new_path):
            DatabaseMigrator(old_path, new_path).update_database()
            os.remove(old_path)

    @staticmethod
    def update_log_database(old_path: str = Config.SQLITE_LOG_DATABASE_PATH) -> None:
        '''直接更新日志数据库'''
        if os.path.isfile(old_path):
            LogDatabaseMigrator(old_path).update_database()

    def check_song_file(self) -> bool:
        '''检查song有关文件并初始化缓存'''
        f = self.check_folder(Config.SONG_FILE_FOLDER_PATH)
        self.logger.info("Initialize song data...")
        try:
            DownloadList.initialize_cache()
            if not Config.SONG_FILE_HASH_PRE_CALCULATE:
                self.logger.info('Song file hash pre-calculate is disabled.')
        except Exception as e:
            self.logger.error(format_exc())
            self.logger.warning('Song data initialization error!')
            f = False
        return f

    def check_content_bundle(self) -> bool:
        '''检查 content bundle 有关文件并初始化缓存'''
        f = self.check_folder(Config.CONTENT_BUNDLE_FOLDER_PATH)
        self.logger.info("Initialize content bundle data...")
        try:
            BundleParser()
        except Exception as e:
            self.logger.error(format_exc())
            self.logger.warning('Content bundle data initialization error!')
            f = False
        return f

    def check_world_map(self) -> bool:
        '''检查 world map 有关文件并初始化缓存'''
        f = self.check_folder(Config.WORLD_MAP_FOLDER_PATH)
        self.logger.info("Initialize world map data...")
        try:
            MapParser()
        except Exception as e:
            self.logger.error(format_exc())
            self.logger.warning('World map data initialization error!')
            f = False
        return f

    def check_before_run(self) -> bool:
        '''运行前检查，返回布尔值'''
        MemoryDatabase()  # 初始化内存数据库
        return self.check_song_file() and self.check_content_bundle() and self.check_update_database() and self.check_world_map()