artemis/core/title.py

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)