Error063/artemis
2
0
Fork 0
forked from Hay1tsme/artemis
Code Pull Requests Activity

add get_allnet_info and config loading safety to all games

Browse Source
This commit is contained in:
Hay1tsme 2023-03-04 21:58:51 -05:00
parent b2b28850dd
commit bfe5294d51
20 changed files with 116 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
core/allnet.py
View File

@ -56,49 +56,7 @@ class AllnetServlet:
self.uri_registry[code] = (uri, host)
else:
for code in mod.game_codes:
if hasattr(mod, "use_default_title") and mod.use_default_title:
if hasattr(mod, "include_protocol") and mod.include_protocol:
if hasattr(mod, "title_secure") and mod.title_secure:
uri = "https://"
else:
uri = "http://"
else:
uri = ""
if core_cfg.server.is_develop:
uri += f"{core_cfg.title.hostname}:{core_cfg.title.port}"
else:
uri += f"{core_cfg.title.hostname}"
uri += f"/{code}/$v"
if hasattr(mod, "trailing_slash") and mod.trailing_slash:
uri += "/"
else:
if hasattr(mod, "uri"):
uri = mod.uri
else:
uri = ""
if hasattr(mod, "host"):
host = mod.host
elif hasattr(mod, "use_default_host") and mod.use_default_host:
if core_cfg.server.is_develop:
host = f"{core_cfg.title.hostname}:{core_cfg.title.port}"
else:
host = f"{core_cfg.title.hostname}"
else:
host = ""
self.uri_registry[code] = (uri, host)
self.logger.warn("Game {_} has no get_allnet_info method.")
self.logger.info(f"Allnet serving {len(self.uri_registry)} games on port {core_cfg.allnet.port}")

9
core/title.py
View File

@ -39,21 +39,22 @@ class TitleServlet():
if hasattr(mod, "game_codes") and hasattr(mod, "index"):
should_call_setup = True
for code in mod.game_codes:
if hasattr(mod.index, "get_allnet_info"):
for code in mod.game_codes:
enabled, _, _ = mod.index.get_allnet_info(code, self.config, self.config_folder)
else:
enabled = True
if enabled:
handler_cls = mod.index(self.config, self.config_folder)
if hasattr(handler_cls, "setup") and should_call_setup:
handler_cls.setup()
should_call_setup = False
self.title_registry[code] = handler_cls
else:
self.logger.warn(f"Game {folder} has no get_allnet_info")
else:
self.logger.error(f"{folder} missing game_code or index in __init__.py")

2
dbutils.py
View File

@ -2,6 +2,7 @@ import yaml
import argparse
from core.config import CoreConfig
from core.data import Data
from os import path
if __name__=='__main__':
parser = argparse.ArgumentParser(description="Database utilities")
@ -16,6 +17,7 @@ if __name__=='__main__':
args = parser.parse_args()
cfg = CoreConfig()
if path.exists(f"{args.config}/core.yaml"):
cfg.update(yaml.safe_load(open(f"{args.config}/core.yaml")))
data = Data(cfg)

1
index.py
View File

@ -90,6 +90,7 @@ if __name__ == "__main__":
exit(1)
cfg: CoreConfig = CoreConfig()
if path.exists(f"{args.config}/core.yaml"):
cfg.update(yaml.safe_load(open(f"{args.config}/core.yaml")))
logger = logging.getLogger("core")

3
read.py
View File

@ -3,7 +3,7 @@ import argparse
import re
import os
import yaml
import importlib
from os import path
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
@ -79,6 +79,7 @@ if __name__ == "__main__":
args = parser.parse_args()
config = CoreConfig()
if path.exists(f"{args.config}/core.yaml"):
config.update(yaml.safe_load(open(f"{args.config}/core.yaml")))
log_fmt_str = "[%(asctime)s] Reader | %(levelname)s | %(message)s"

8
titles/chuni/__init__.py
View File

@ -6,13 +6,5 @@ from titles.chuni.read import ChuniReader
index = ChuniServlet
database = ChuniData
reader = ChuniReader
use_default_title = True
include_protocol = True
title_secure = False
game_codes = [ChuniConstants.GAME_CODE, ChuniConstants.GAME_CODE_NEW]
trailing_slash = True
use_default_host = False
host = ""
current_schema_version = 1

2
titles/chuni/const.py
View File

@ -2,6 +2,8 @@ class ChuniConstants():
GAME_CODE = "SDBT"
GAME_CODE_NEW = "SDHD"
CONFIG_NAME = "chuni.yaml"
VER_CHUNITHM = 0
VER_CHUNITHM_PLUS = 1
VER_CHUNITHM_AIR = 2

19
titles/chuni/index.py
View File

@ -8,6 +8,8 @@ import inflection
import string
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from os import path
from typing import Tuple
from core import CoreConfig
from titles.chuni.config import ChuniConfig
@ -30,7 +32,8 @@ class ChuniServlet():
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
self.game_cfg = ChuniConfig()
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/chuni.yaml")))
if path.exists(f"{cfg_dir}/{ChuniConstants.CONFIG_NAME}"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{ChuniConstants.CONFIG_NAME}")))
self.versions = [
ChuniBase(core_cfg, self.game_cfg),
@ -68,6 +71,20 @@ class ChuniServlet():
coloredlogs.install(level=self.game_cfg.server.loglevel, logger=self.logger, fmt=log_fmt_str)
self.logger.inited = True
@classmethod
def get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = ChuniConfig()
if path.exists(f"{cfg_dir}/{ChuniConstants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{ChuniConstants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")
if core_cfg.server.is_develop:
return (True, f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v/", "")
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v/", "")
def render_POST(self, request: Request, version: int, url_path: str) -> bytes:
req_raw = request.content.getvalue()
url_split = url_path.split("/")

11
titles/cxb/__init__.py
View File

@ -6,16 +6,5 @@ from titles.cxb.read import CxbReader
index = CxbServlet
database = CxbData
reader = CxbReader
use_default_title = False
include_protocol = True
title_secure = True
game_codes = [CxbConstants.GAME_CODE]
trailing_slash = True
use_default_host = False
include_port = True
uri = "http://$h:$p/" # If you care about the allnet response you're probably running with no SSL
host = ""
current_schema_version = 1

20
titles/cxb/index.py
View File

@ -7,7 +7,8 @@ import re
import inflection
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
from typing import Dict
from typing import Dict, Tuple
from os import path
from core.config import CoreConfig
from titles.cxb.config import CxbConfig
@ -22,7 +23,8 @@ class CxbServlet(resource.Resource):
self.cfg_dir = cfg_dir
self.core_cfg = core_cfg
self.game_cfg = CxbConfig()
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/cxb.yaml")))
if path.exists(f"{cfg_dir}/{CxbConstants.CONFIG_NAME}"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{CxbConstants.CONFIG_NAME}")))
self.logger = logging.getLogger("cxb")
if not hasattr(self.logger, "inited"):
@ -49,6 +51,20 @@ class CxbServlet(resource.Resource):
CxbRevSunriseS2(core_cfg, self.game_cfg),
]
@classmethod
def get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = CxbConfig()
if path.exists(f"{cfg_dir}/{CxbConstants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{CxbConstants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")
if core_cfg.server.is_develop:
return (True, f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v/", "")
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v/", "")
def setup(self):
if self.game_cfg.server.enable:
endpoints.serverFromString(reactor, f"tcp:{self.game_cfg.server.port}:interface={self.core_cfg.server.listen_address}")\

8
titles/diva/__init__.py
View File

@ -6,13 +6,5 @@ from titles.diva.read import DivaReader
index = DivaServlet
database = DivaData
reader = DivaReader
use_default_title = True
include_protocol = True
title_secure = False
game_codes = [DivaConstants.GAME_CODE]
trailing_slash = True
use_default_host = False
host = ""
current_schema_version = 1

2
titles/diva/const.py
View File

@ -1,6 +1,8 @@
class DivaConstants():
GAME_CODE = "SBZV"
CONFIG_NAME = "diva.yaml"
VER_PROJECT_DIVA_ARCADE = 0
VER_PROJECT_DIVA_ARCADE_FUTURE_TONE = 1

20
titles/diva/index.py
View File

@ -6,16 +6,20 @@ import zlib
import json
import urllib.parse
import base64
from os import path
from typing import Tuple
from core.config import CoreConfig
from titles.diva.config import DivaConfig
from titles.diva.const import DivaConstants
from titles.diva.base import DivaBase
class DivaServlet():
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
self.game_cfg = DivaConfig()
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/diva.yaml")))
if path.exists(f"{cfg_dir}/{DivaConstants.CONFIG_NAME}"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{DivaConstants.CONFIG_NAME}")))
self.base = DivaBase(core_cfg, self.game_cfg)
@ -36,6 +40,20 @@ class DivaServlet():
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 get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = DivaConfig()
if path.exists(f"{cfg_dir}/{DivaConstants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{DivaConstants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")
if core_cfg.server.is_develop:
return (True, f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v/", "")
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v/", "")
def render_POST(self, req: Request, version: int, url_path: str) -> bytes:
req_raw = req.content.getvalue()
url_header = req.getAllHeaders()

8
titles/mai2/__init__.py
View File

@ -6,13 +6,5 @@ from titles.mai2.read import Mai2Reader
index = Mai2Servlet
database = Mai2Data
reader = Mai2Reader
use_default_title = True
include_protocol = True
title_secure = False
game_codes = [Mai2Constants.GAME_CODE]
trailing_slash = True
use_default_host = False
host = ""
current_schema_version = 2

18
titles/mai2/index.py
View File

@ -6,6 +6,8 @@ import string
import logging, coloredlogs
import zlib
from logging.handlers import TimedRotatingFileHandler
from os import path
from typing import Tuple
from core.config import CoreConfig
from titles.mai2.config import Mai2Config
@ -22,6 +24,7 @@ class Mai2Servlet():
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
self.game_cfg = Mai2Config()
if path.exists(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}")))
self.versions = [
@ -50,6 +53,21 @@ class Mai2Servlet():
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 get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = Mai2Config()
if path.exists(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")
if core_cfg.server.is_develop:
return (True, f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v/", f"{core_cfg.title.hostname}:{core_cfg.title.port}/")
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v/", f"{core_cfg.title.hostname}/")
def render_POST(self, request: Request, version: int, url_path: str) -> bytes:
req_raw = request.content.getvalue()
url = request.uri.decode()

2
titles/ongeki/const.py
View File

@ -3,6 +3,8 @@ from enum import Enum
class OngekiConstants():
GAME_CODE = "SDDT"
CONFIG_NAME = "ongeki.yaml"
VER_ONGEKI = 0
VER_ONGEKI_PLUS = 1
VER_ONGEKI_SUMMER = 2

8
titles/ongeki/index.py
View File

@ -25,8 +25,8 @@ class OngekiServlet():
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
self.game_cfg = OngekiConfig()
if path.exists(f"{cfg_dir}/ongeki.yaml"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/ongeki.yaml")))
if path.exists(f"{cfg_dir}/{OngekiConstants.CONFIG_NAME}"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{OngekiConstants.CONFIG_NAME}")))
self.versions = [
OngekiBase(core_cfg, self.game_cfg),
@ -59,7 +59,9 @@ class OngekiServlet():
@classmethod
def get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = OngekiConfig()
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/ongeki.yaml")))
if path.exists(f"{cfg_dir}/{OngekiConstants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{OngekiConstants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")

5
titles/pokken/index.py
View File

@ -12,6 +12,7 @@ from google.protobuf.message import DecodeError
from core.config import CoreConfig
from titles.pokken.config import PokkenConfig
from titles.pokken.base import PokkenBase
from titles.pokken.const import PokkenConstants
class PokkenServlet(resource.Resource):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
@ -46,7 +47,9 @@ class PokkenServlet(resource.Resource):
@classmethod
def get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = PokkenConfig()
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/pokken.yaml")))
if path.exists(f"{cfg_dir}/{PokkenConstants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{PokkenConstants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")

8
titles/wacca/__init__.py
View File

@ -8,13 +8,5 @@ index = WaccaServlet
database = WaccaData
reader = WaccaReader
frontend = WaccaFrontend
use_default_title = True
include_protocol = True
title_secure = False
game_codes = [WaccaConstants.GAME_CODE]
trailing_slash = False
use_default_host = False
host = ""
current_schema_version = 3

11
titles/wacca/index.py
View File

@ -24,8 +24,8 @@ class WaccaServlet():
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
self.game_cfg = WaccaConfig()
if path.exists(f"{cfg_dir}/wacca.yaml"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/wacca.yaml")))
if path.exists(f"{cfg_dir}/{WaccaConstants.CONFIG_NAME}"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{WaccaConstants.CONFIG_NAME}")))
self.versions = [
WaccaBase(core_cfg, self.game_cfg),
@ -55,15 +55,16 @@ class WaccaServlet():
@classmethod
def get_allnet_info(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str, str]:
game_cfg = WaccaConfig()
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/wacca.yaml")))
if path.exists(f"{cfg_dir}/{WaccaConstants.CONFIG_NAME}"):
game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/{WaccaConstants.CONFIG_NAME}")))
if not game_cfg.server.enable:
return (False, "", "")
if core_cfg.server.is_develop:
return (True, f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v/", "")
return (True, f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v", "")
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v/", "")
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v", "")
def render_POST(self, request: Request, version: int, url_path: str) -> bytes:
def end(resp: Dict) -> bytes: