1
0
forked from Hay1tsme/artemis
artemis/core/title.py

179 lines
7.2 KiB
Python
Raw Normal View History

from typing import Dict, List, Tuple
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
2023-02-16 22:13:41 +00:00
from twisted.web.http import Request
from core.config import CoreConfig
from core.data import Data
2023-02-16 22:13:41 +00:00
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
2023-03-09 16:38:58 +00:00
class TitleServlet:
title_registry: Dict[str, BaseServlet] = {}
2023-03-09 16:38:58 +00:00
def __init__(self, core_cfg: CoreConfig, cfg_folder: str):
super().__init__()
self.config = core_cfg
self.config_folder = cfg_folder
2023-02-16 22:13:41 +00:00
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"
2023-03-09 16:38:58 +00:00
log_fmt = logging.Formatter(log_fmt_str)
2023-02-16 22:13:41 +00:00
2023-03-09 16:38:58 +00:00
fileHandler = TimedRotatingFileHandler(
"{0}/{1}.log".format(self.config.server.log_dir, "title"),
when="d",
backupCount=10,
)
2023-02-16 22:13:41 +00:00
fileHandler.setFormatter(log_fmt)
2023-03-09 16:38:58 +00:00
2023-02-16 22:13:41 +00:00
consoleHandler = logging.StreamHandler()
consoleHandler.setFormatter(log_fmt)
self.logger.addHandler(fileHandler)
self.logger.addHandler(consoleHandler)
2023-03-09 16:38:58 +00:00
2023-02-16 22:13:41 +00:00
self.logger.setLevel(core_cfg.title.loglevel)
2023-03-09 16:38:58 +00:00
coloredlogs.install(
level=core_cfg.title.loglevel, logger=self.logger, fmt=log_fmt_str
)
2023-02-16 22:13:41 +00:00
self.logger.initialized = True
2023-03-09 16:38:58 +00:00
2023-02-18 05:00:30 +00:00
plugins = Utils.get_all_titles()
2023-03-09 16:38:58 +00:00
2023-02-18 05:00:30 +00:00
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)
2023-03-09 16:38:58 +00:00
if hasattr(handler_cls, "setup") and should_call_setup:
handler_cls.setup()
should_call_setup = False
2023-03-09 16:38:58 +00:00
self.title_registry[code] = handler_cls
2023-03-09 16:38:58 +00:00
else:
self.logger.error(f"{folder} missing game_code or index in __init__.py, or is_game_enabled in index")
2023-03-09 16:38:58 +00:00
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 ''}"
2023-03-09 16:38:58 +00:00
)
def render_GET(self, request: Request, endpoints: dict) -> bytes:
code = endpoints["title"]
subaction = endpoints['subaction']
if code not in self.title_registry:
2023-08-08 14:17:56 +00:00
self.logger.warning(f"Unknown game code {code}")
2023-03-02 05:14:13 +00:00
request.setResponseCode(404)
return b""
2023-03-09 16:38:58 +00:00
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)
2023-03-09 16:38:58 +00:00
def render_POST(self, request: Request, endpoints: dict) -> bytes:
code = endpoints["title"]
subaction = endpoints['subaction']
if code not in self.title_registry:
2023-08-08 14:17:56 +00:00
self.logger.warning(f"Unknown game code {code}")
2023-03-02 05:14:13 +00:00
request.setResponseCode(404)
return b""
2023-03-09 16:38:58 +00:00
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)