diff --git a/game_server/config/__init__.py b/game_server/config/__init__.py deleted file mode 100644 index 9254121..0000000 --- a/game_server/config/__init__.py +++ /dev/null @@ -1,57 +0,0 @@ -# 头文件, 负责统一引用外部库 -# 本文件夹下的文件禁止引用此文件 -import os -import sys -import json -import time -import zlib -import base64 -import socket -import struct -import logging -import requests -import threading -from dynaconf import Dynaconf -from flask import Flask, request, jsonify, send_from_directory, Blueprint -from pathlib import Path -from enum import Enum -from .config import * -from .log import * # TODO: After config to tempfix config.json not found error -from lib import proto as protos -from pprint import pprint -import importlib -import re - - -__all__ = [ - "os", - "sys", - "json", - "time", - "zlib", - "base64", - "socket", - "struct", - "logging", - "threading", - "Dynaconf", - "Flask", - "request", - "requests", - "jsonify", - "send_from_directory", - "Blueprint", - "Path", - "Enum", - "Log", - "Error", - "Warn", - "Info", - "Debug", - "Config", - "Root", - "protos", - "pprint", - "importlib", - "re" -] diff --git a/game_server/config/config.py b/game_server/config/config.py deleted file mode 100644 index c23022d..0000000 --- a/game_server/config/config.py +++ /dev/null @@ -1,42 +0,0 @@ -import os -import json -from dynaconf import Dynaconf - -# 此处设置默认值 -Root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -ConfigData = { - "LogLevel": "INFO", - "MaxSessions": 10, - "GameServer": { - "Ip": "127.0.0.1", - "Port": 16100 - }, - "SdkServer": { - "Ip": "127.0.0.1", - "Port": 80 - }, - "VerboseLevel":1, - "RegionName":"MikuBH3", - "UseLocalCache":False -} - -class ConfigInit: - def __init__(self): - self.ConfigPath = f"{Root}/../Config.json" - self.ConfigData = self.Load() - - if not os.path.exists(self.ConfigPath): - with open(self.ConfigPath, 'w') as f: - json.dump(ConfigData, f, indent=4) - - def Load(self): - Settings = Dynaconf( - settings_files = [self.ConfigPath], - default_settings = ConfigData - ) - return Settings - - def Get(self): - return self.ConfigData - -Config = ConfigInit().Load() \ No newline at end of file diff --git a/game_server/resource/__init__.py b/game_server/resource/__init__.py index e03353a..e28fe79 100644 --- a/game_server/resource/__init__.py +++ b/game_server/resource/__init__.py @@ -2,17 +2,19 @@ import json import traceback from typing import Dict, Type, TypeVar, Optional, Any, List -from game_server.config.log import Error, Info +from utils.logger import Error, Info from game_server.resource.base_resource import BaseResource from game_server.resource.decorators import resource_registry import game_server.resource.configdb # noqa: F401 T = TypeVar("T", bound=BaseResource) + def filter_data(cls: Type, data): valid_fields = cls.__annotations__.keys() return {field: data.get(field, None) for field in valid_fields} + class ResourceManager: def __init__(self): self.data: Dict[Type[T], Dict[Any, T]] = {} @@ -57,4 +59,5 @@ class ResourceManager: def instance(): return resource_manager + resource_manager = ResourceManager() diff --git a/utils/aes.py b/utils/aes.py new file mode 100644 index 0000000..5fed06e --- /dev/null +++ b/utils/aes.py @@ -0,0 +1,16 @@ +import base64 +from Crypto.Cipher import AES +from Crypto.Util.Padding import pad, unpad + + +def encrypt_ecb(key: str, data: str): + cipher = AES.new(bytes.fromhex(key.replace(" ", "")), AES.MODE_ECB) + encrypted = cipher.encrypt(pad(data.encode(), AES.block_size)) + return base64.b64encode(encrypted).decode() + + +def decrypt_ecb(key: str, data: str): + data = base64.b64decode(data) + cipher = AES.new(bytes.fromhex(key.replace(" ", "")), AES.MODE_ECB) + decrypted = cipher.decrypt(data) + return unpad(decrypted, AES.block_size).decode() diff --git a/utils/config.py b/utils/config.py new file mode 100644 index 0000000..2071d48 --- /dev/null +++ b/utils/config.py @@ -0,0 +1,54 @@ +from dataclasses import asdict, dataclass +import os +import json +from dacite import from_dict + + +@dataclass +class ServerConfig: + IP: str + Port: int + + +@dataclass +class ConfigData: + LogLevel: str + GameServer: ServerConfig + SDKServer: ServerConfig + VerboseLevel: int + RegionName: str + UseLocalCache: bool + AESKeys: dict[str, str] + EnableDispatchEncryption: bool + + def write_default_config(): + config = ConfigData( + LogLevel="INFO", + GameServer=ServerConfig(IP="127.0.0.1", Port=16100), + SDKServer=ServerConfig(IP="127.0.0.1", Port=80), + VerboseLevel=1, + RegionName="MikuBH3", + UseLocalCache=False, + EnableDispatchEncryption=True, + AESKeys={ + "7.9.0_gf_pc": "36 31 65 37 64 33 65 66 33 32 30 63 31 35 66 66 61 64 37 61 66 32 31 34 61 64 65 64 32 34 33 38", + "7.8.0_os_pc": "64 34 32 33 30 30 31 62 32 36 38 34 62 33 62 30 61 33 30 38 66 37 65 35 63 30 61 38 66 33 65 32", + }, + ) + with open("Config.json", "w") as f: + f.write(json.dumps(asdict(config), indent=2)) + + return config + + def load(): + if not os.path.exists("Config.json"): + return ConfigData.write_default_config() + + with open("Config.json", "r", encoding="utf-8") as f: + try: + return from_dict(ConfigData, json.load(f)) + except Exception: + return ConfigData.write_default_config() + + +Config = ConfigData.load() diff --git a/game_server/config/log.py b/utils/logger.py similarity index 68% rename from game_server/config/log.py rename to utils/logger.py index 0a84b53..7021df7 100644 --- a/game_server/config/log.py +++ b/utils/logger.py @@ -1,37 +1,39 @@ -import json import sys from loguru import logger +from utils.config import Config -# Configuration for the logger logger.remove() -with open("Config.json", "r", encoding="utf-8") as f: - LogLevel = json.load(f)["LogLevel"] + LevelList = ["ERROR", "WARNING", "INFO", "DEBUG"] -CodeColorDict = { - "ERROR": "red", - "WARNING": "yellow", - "INFO": "green", - "DEBUG": "blue" -} +CodeColorDict = {"ERROR": "red", "WARNING": "yellow", "INFO": "green", "DEBUG": "blue"} + def custom_format(record): color = CodeColorDict[record["level"].name] return f"<{color}>{record['level'].name} : {record['message']}\n" -logger.add(sys.stdout, format=custom_format, colorize=True, level=LogLevel) + +logger.add(sys.stdout, format=custom_format, colorize=True, level=Config.LogLevel) + def Log(msg, types): - if types in CodeColorDict and LevelList.index(types) <= LevelList.index(LogLevel): + if types in CodeColorDict and LevelList.index(types) <= LevelList.index( + Config.LogLevel + ): getattr(logger, types.lower())(msg) + def Error(msg): Log(msg, "ERROR") + def Warn(msg): Log(msg, "WARNING") + def Info(msg): Log(msg, "INFO") + def Debug(msg): Log(msg, "DEBUG")