3
2
forked from Dniel97/artemis
artemis/core/title.py
Kevin Trocolli cb8eaae2c0 Per-version URI/Host (#66)
Allows setting allnet uri/host response based on things like version, config files, and other factors to accommodate a wider range of potential setups under the same roof. This DOES require all titles to adopt a new structure but it's documented and should hopefully be somewhat intuitive.

Co-authored-by: Hay1tsme <kevin@hay1ts.me>
Reviewed-on: Hay1tsme/artemis#66
Co-authored-by: Kevin Trocolli <pitok236@gmail.com>
Co-committed-by: Kevin Trocolli <pitok236@gmail.com>
2023-11-09 02:17:48 +00:00

179 lines
7.2 KiB
Python

from typing import Dict, List, Tuple
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
from twisted.web.http import Request
from core.config import CoreConfig
from core.data import Data
from core.utils import Utils
class BaseServlet:
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
self.game_cfg = None
self.logger = logging.getLogger("title")
@classmethod
def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool:
"""Called during boot to check if a specific game code should load.
Args:
game_code (str): 4 character game code
core_cfg (CoreConfig): CoreConfig class
cfg_dir (str): Config directory
Returns:
bool: True if the game is enabled and set to run, False otherwise
"""
return False
def get_endpoint_matchers(self) -> Tuple[List[Tuple[str, str, Dict]], List[Tuple[str, str, Dict]]]:
"""Called during boot to get all matcher endpoints this title servlet handles
Returns:
Tuple[List[Tuple[str, str, Dict]], List[Tuple[str, str, Dict]]]: A 2-length tuple where offset 0 is GET and offset 1 is POST,
containing a list of 3-length tuples where offset 0 is the name of the function in the handler that should be called, offset 1
is the matching string, and offset 2 is a dict containing rules for the matcher.
"""
return (
[("render_GET", "/{game}/{version}/{endpoint}", {'game': R'S...'})],
[("render_POST", "/{game}/{version}/{endpoint}", {'game': R'S...'})]
)
def setup(self) -> None:
"""Called once during boot, should contain any additional setup the handler must do, such as starting any sub-services
"""
pass
def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
"""Called any time a request to PowerOn is made to retrieve the url/host strings to be sent back to the game
Args:
game_code (str): 4 character game code
game_ver (int): version, expressed as an integer by multiplying by 100 (1.10 -> 110)
keychip (str): Keychip serial of the requesting machine, can be used to deliver specific URIs to different machines
Returns:
Tuple[str, str]: A tuple where offset 0 is the allnet uri field, and offset 1 is the allnet host field
"""
if not self.core_cfg.server.is_using_proxy and Utils.get_title_port(self.core_cfg) != 80:
return (f"http://{self.core_cfg.title.hostname}:{Utils.get_title_port(self.core_cfg)}/{game_code}/{game_ver}/", "")
return (f"http://{self.core_cfg.title.hostname}/{game_code}/{game_ver}/", "")
def get_mucha_info(self, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str]:
"""Called once during boot to check if this game is a mucha game
Args:
core_cfg (CoreConfig): CoreConfig class
cfg_dir (str): Config directory
Returns:
Tuple[bool, str]: Tuple where offset 0 is true if the game is enabled, false otherwise, and offset 1 is the game CD
"""
return (False, "")
def render_POST(self, request: Request, game_code: str, matchers: Dict) -> bytes:
self.logger.warn(f"{game_code} Does not dispatch POST")
return None
def render_GET(self, request: Request, game_code: str, matchers: Dict) -> bytes:
self.logger.warn(f"{game_code} Does not dispatch GET")
return None
class TitleServlet:
title_registry: Dict[str, BaseServlet] = {}
def __init__(self, core_cfg: CoreConfig, cfg_folder: str):
super().__init__()
self.config = core_cfg
self.config_folder = cfg_folder
self.data = Data(core_cfg)
self.logger = logging.getLogger("title")
if not hasattr(self.logger, "initialized"):
log_fmt_str = "[%(asctime)s] Title | %(levelname)s | %(message)s"
log_fmt = logging.Formatter(log_fmt_str)
fileHandler = TimedRotatingFileHandler(
"{0}/{1}.log".format(self.config.server.log_dir, "title"),
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(core_cfg.title.loglevel)
coloredlogs.install(
level=core_cfg.title.loglevel, logger=self.logger, fmt=log_fmt_str
)
self.logger.initialized = True
plugins = Utils.get_all_titles()
for folder, mod in plugins.items():
if hasattr(mod, "game_codes") and hasattr(mod, "index") and hasattr(mod.index, "is_game_enabled"):
should_call_setup = True
game_servlet: BaseServlet = mod.index
game_codes: List[str] = mod.game_codes
for code in game_codes:
if game_servlet.is_game_enabled(code, self.config, self.config_folder):
handler_cls = game_servlet(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.error(f"{folder} missing game_code or index in __init__.py, or is_game_enabled in index")
self.logger.info(
f"Serving {len(self.title_registry)} game codes {'on port ' + str(core_cfg.title.port) if core_cfg.title.port > 0 else ''}"
)
def render_GET(self, request: Request, endpoints: dict) -> bytes:
code = endpoints["title"]
subaction = endpoints['subaction']
if code not in self.title_registry:
self.logger.warning(f"Unknown game code {code}")
request.setResponseCode(404)
return b""
index = self.title_registry[code]
handler = getattr(index, f"{subaction}", None)
if handler is None:
self.logger.error(f"{code} does not have handler for GET subaction {subaction}")
request.setResponseCode(500)
return b""
return handler(request, code, endpoints)
def render_POST(self, request: Request, endpoints: dict) -> bytes:
code = endpoints["title"]
subaction = endpoints['subaction']
if code not in self.title_registry:
self.logger.warning(f"Unknown game code {code}")
request.setResponseCode(404)
return b""
index = self.title_registry[code]
handler = getattr(index, f"{subaction}", None)
if handler is None:
self.logger.error(f"{code} does not have handler for POST subaction {subaction}")
request.setResponseCode(500)
return b""
endpoints.pop("title")
endpoints.pop("subaction")
return handler(request, code, endpoints)