refactor: Refactor config

game_server/config/__init__.py

@@ -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"
-]

game_server/config/config.py

@@ -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()

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()

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()

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()

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}</{color}> : {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")