添加 example_config/fgoa.yaml

example_config/fgoa.yaml

@@ -0,0 +1,3 @@
+server:
+  enable: True
+  loglevel: "info"

titles/FGOA/__init__.py

@@ -0,0 +1,5 @@
+from titles.fgoa.index import FGOAServlet
+from titles.fgoa.const import FGOAConstants
+
+index = FGOAServlet
+game_codes = [FGOAConstants.GAME_CODE]

titles/FGOA/base.py

@@ -0,0 +1,30 @@
+from datetime import date, datetime, timedelta
+from typing import Any, Dict, List
+import json
+import logging
+from enum import Enum
+
+from core.config import CoreConfig
+from titles.fgoa.config import FGOAConfig
+from titles.fgoa.const import FGOAConstants
+
+
+class FGOABase:
+    def __init__(self, core_cfg: CoreConfig, game_cfg: FGOAConfig) -> None:
+        self.core_cfg = core_cfg
+        self.game_config = game_cfg
+        self.date_time_format = "%Y-%m-%d %H:%M:%S"
+        self.date_time_format_ext = (
+            "%Y-%m-%d %H:%M:%S.%f"  # needs to be lopped off at [:-5]
+        )
+        self.date_time_format_short = "%Y-%m-%d"
+        self.logger = logging.getLogger("fgoa")
+        self.game = FGOAConstants.GAME_CODE
+        self.version = FGOAConstants.VER_FGOA_SEASON_1
+
+    @staticmethod
+    def _parse_int_ver(version: str) -> str:
+        return version.replace(".", "")[:3]
+
+    async def handle_game_init_request(self, data: Dict) -> Dict:
+        return f""

titles/FGOA/config.py

@@ -0,0 +1,24 @@
+from core.config import CoreConfig
+
+
+class FGOAServerConfig:
+    def __init__(self, parent: "FGOAConfig") -> None:
+        self.__config = parent
+
+    @property
+    def enable(self) -> bool:
+        return CoreConfig.get_config_field(
+            self.__config, "fgo", "server", "enable", default=True
+        )
+
+    @property
+    def loglevel(self) -> int:
+        return CoreConfig.str_to_loglevel(
+            CoreConfig.get_config_field(
+                self.__config, "fgo", "server", "loglevel", default="info"
+            )
+        )
+
+class FGOAConfig(dict):
+    def __init__(self) -> None:
+        self.server = FGOAServerConfig(self)

titles/FGOA/const.py

@@ -0,0 +1,14 @@
+class FGOAConstants():
+    GAME_CODE = "SDEJ"
+
+    CONFIG_NAME = "fgoa.yaml"
+
+    VER_FGOA_SEASON_1 = 0
+
+    VERSION_STRING = (
+        "Fate/Grand Order Arcade",
+    )
+
+    @classmethod
+    def game_ver_to_string(cls, ver: int):
+        return cls.VERSION_STRING[ver]

titles/FGOA/index.py

@@ -0,0 +1,123 @@
+import json
+import inflection
+import yaml
+import string
+import logging
+import coloredlogs
+import zlib
+import base64
+import urllib.parse
+
+from os import path
+from typing import Dict, List, Tuple
+from logging.handlers import TimedRotatingFileHandler
+
+from starlette.routing import Route
+from starlette.responses import Response
+from starlette.requests import Request
+from starlette.responses import PlainTextResponse
+
+from core.config import CoreConfig
+from core.title import BaseServlet
+from core.utils import Utils
+
+from titles.fgoa.base import FGOABase
+from titles.fgoa.config import FGOAConfig
+from titles.fgoa.const import FGOAConstants
+
+class FGOAServlet(BaseServlet):
+    def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
+        self.core_cfg = core_cfg
+        self.game_cfg = FGOAConfig()
+        if path.exists(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}"):
+            self.game_cfg.update(
+                yaml.safe_load(open(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}"))
+            )
+
+        self.versions = [
+            FGOABase(core_cfg, self.game_cfg),
+        ]
+
+        self.logger = logging.getLogger("fgoa")
+        log_fmt_str = "[%(asctime)s] FGOA | %(levelname)s | %(message)s"
+        log_fmt = logging.Formatter(log_fmt_str)
+        fileHandler = TimedRotatingFileHandler(
+            "{0}/{1}.log".format(self.core_cfg.server.log_dir, "fgoa"),
+            encoding="utf8",
+            when="d",
+            backupCount=10,
+        )
+
+        fileHandler.setFormatter(log_fmt)
+
+        consoleHandler = logging.StreamHandler()
+        consoleHandler.setFormatter(log_fmt)
+
+        self.logger.addHandler(fileHandler)
+        self.logger.addHandler(consoleHandler)
+
+        self.logger.setLevel(self.game_cfg.server.loglevel)
+        coloredlogs.install(
+            level=self.game_cfg.server.loglevel, logger=self.logger, fmt=log_fmt_str
+        )
+
+    @classmethod
+    def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool:
+        game_cfg = FGOAConfig()
+
+        if path.exists(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}"):
+            game_cfg.update(
+                yaml.safe_load(open(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}"))
+            )
+
+        if not game_cfg.server.enable:
+            return False
+        
+        return True
+    
+    def get_routes(self) -> List[Route]:
+        return [
+            Route("/SDEJ/{version:int}/{endpoint:str}", self.render_POST, methods=['POST'])
+        ]
+
+    def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
+        if not self.core_cfg.server.is_using_proxy and Utils.get_title_port(self.core_cfg) != 80:
+            return (f"http://{self.core_cfg.server.hostname}:{Utils.get_title_port(self.core_cfg)}/{game_code}/{game_ver}", self.core_cfg.server.hostname)
+        
+        return (f"http://{self.core_cfg.server.hostname}/{game_code}/{game_ver}", self.core_cfg.server.hostname)
+    
+
+    async def render_POST(self, request: Request) -> bytes:
+        version: int = request.path_params.get('version')
+        endpoint: str = request.path_params.get('endpoint')
+        req_raw = await request.body()
+        internal_ver = 0
+        client_ip = Utils.get_ip_addr(request)
+
+        if all(c in string.hexdigits for c in endpoint) and len(endpoint) == 32:
+            # If we get a 32 character long hex string, it's a hash and we're
+            # doing encrypted. The likelyhood of false positives is low but
+            # technically not 0
+            self.logger.error("Encryption not supported at this time")
+
+        self.logger.debug(req_raw)
+        
+        self.logger.info(f"v{version} {endpoint} request from {client_ip}")
+
+        func_to_find = "handle_" + inflection.underscore(endpoint) + "_request"
+
+        try:
+            handler = getattr(self.versions[internal_ver], func_to_find)
+            resp = await handler(req_raw)
+
+        except Exception as e:
+            self.logger.error(f"Error handling v{version} method {endpoint} - {e}")
+            raise
+            return Response(zlib.compress(b'{"stat": "0"}'))
+
+        if resp is None:
+            resp = {"returnCode": 1}
+
+        self.logger.debug(f"Response {resp}")
+
+        return Response(zlib.compress(json.dumps(resp, ensure_ascii=False).encode("utf-8")))