import functools
import json
import re
from abc import ABC
from typing import Optional, List
import parse
from typing_extensions import override
from mcdreforged.handler.abstract_server_handler import AbstractServerHandler
from mcdreforged.info_reactor.info import Info
from mcdreforged.info_reactor.server_information import ServerInformation
from mcdreforged.minecraft.rtext.text import RTextBase
from mcdreforged.plugin.meta.version import VersionParsingError
from mcdreforged.utils import string_util
from mcdreforged.utils.types.message import MessageText
[文档]
class AbstractMinecraftHandler(AbstractServerHandler, ABC):
"""
An abstract handler for Minecraft Java Edition servers
"""
@override
def get_stop_command(self) -> str:
return 'stop'
@classmethod
@functools.lru_cache()
def __get_player_message_parsers(cls) -> List[parse.Parser]:
"""
The return value is cached for reuse. Do not modify
"""
formatters = cls.get_player_message_parsing_formatter()
return list(map(parse.Parser, formatters))
@override
def get_send_message_command(self, target: str, message: MessageText, server_information: ServerInformation) -> Optional[str]:
can_do_execute = False
if server_information.version is not None:
try:
from mcdreforged.plugin.meta.version import Version
version = Version(server_information.version.split(' ')[0])
if version >= Version('1.13.0'):
can_do_execute = True
except VersionParsingError:
pass
command = 'tellraw {} {}'.format(target, self.format_message(message))
if can_do_execute:
command = 'execute at @p run ' + command
return command
@override
def get_broadcast_message_command(self, message: MessageText, server_information: ServerInformation) -> Optional[str]:
return self.get_send_message_command('@a', message, server_information)
@classmethod
@override
def _get_server_stdout_raw_result(cls, text: str) -> Info:
raw_result = super()._get_server_stdout_raw_result(text)
# Minecraft <= 1.12.x might output minecraft color codes to the console
# Just remove that
raw_result.content = string_util.clean_minecraft_color_code(raw_result.content)
return raw_result
__player_name_regex = re.compile(r'[a-zA-Z0-9_]{3,16}')
@classmethod
def _verify_player_name(cls, name: str):
return cls.__player_name_regex.fullmatch(name) is not None
@override
def parse_server_stdout(self, text: str):
result = super().parse_server_stdout(text)
for parser in self.__get_player_message_parsers():
parsed = parser.parse(result.content)
if parsed is not None and self._verify_player_name(parsed['name']):
result.player, result.content = parsed['name'], parsed['message']
break
return result
__player_joined_parser = parse.Parser('{name}[{}] logged in with entity id {} at ({})')
__player_left_regex = re.compile(r'[a-zA-Z0-9_]{3,16} left the game')
__server_version_parser = parse.Parser('Starting minecraft server version {version}')
__server_address_parser = parse.Parser('Starting Minecraft server on {}:{:d}')
__server_startup_done_regex = re.compile(r'Done \([0-9.]*s\)! For help, type "help"( or "\?")?')
__rcon_started_regex = re.compile(r'RCON running on [\w.]+:\d+')
@override
def parse_player_joined(self, info: Info):
# Steve[/127.0.0.1:9864] logged in with entity id 131 at (187.2703, 146.79014, 404.84718)
if not info.is_user:
parsed = self.__player_joined_parser.parse(info.content)
if parsed is not None and self._verify_player_name(parsed['name']):
return parsed['name']
return None
@override
def parse_player_left(self, info: Info):
# Steve left the game
if not info.is_user and self.__player_left_regex.fullmatch(info.content):
return info.content.split(' ')[0]
return None
@override
def parse_server_version(self, info: Info):
if not info.is_user:
parsed = self.__server_version_parser.parse(info.content)
if parsed is not None:
return parsed['version']
return None
@override
def parse_server_address(self, info: Info):
if not info.is_user:
parsed = self.__server_address_parser.parse(info.content)
if parsed is not None:
return parsed[0], parsed[1]
return None
@override
def test_server_startup_done(self, info: Info):
# 1.13+ Done (3.500s)! For help, type "help"
# 1.13- Done (3.500s)! For help, type "help" or "?"
return info.is_from_server and self.__server_startup_done_regex.fullmatch(info.content) is not None
@override
def test_rcon_started(self, info: Info):
# RCON running on 0.0.0.0:25575
return info.is_from_server and self.__rcon_started_regex.fullmatch(info.content) is not None
@override
def test_server_stopping(self, info: Info):
# Stopping server
return info.is_from_server and info.content == 'Stopping server'