您最多选择25个主题 主题必须以字母或数字开头,可以包含连字符 (-),并且长度不得超过35个字符

198 行
7.9 KiB

  1. import os
  2. import asyncio
  3. import importlib
  4. from typing import Dict, Optional, List, Union
  5. from logging.config import dictConfig
  6. from logging import getLogger, Logger
  7. from ..variables.loader import Datavars
  8. from ..commands.commands import Command, CommandRunner
  9. from .utils.responses import ResponseStructure
  10. # from .utils.workers import Worker
  11. from .schemas.config import ConfigSchema
  12. from calculate.utils.tools import Singleton
  13. # Получаем конфигурацию сервера.
  14. TESTING = bool(os.environ.get("TESTING", False))
  15. if not TESTING:
  16. from .config import config
  17. server_config = ConfigSchema(**config)
  18. else:
  19. from tests.server.config import config
  20. server_config = ConfigSchema(**config)
  21. class ServerData(metaclass=Singleton):
  22. def __init__(self, config: ConfigSchema = server_config):
  23. self.event_loop = asyncio.get_event_loop()
  24. self._variables_path = config.variables_path
  25. # Конфигурируем логгирование.
  26. dictConfig(config.logger_config)
  27. self.logger: Logger = getLogger("main")
  28. self.log_message = {'DEBUG': self.logger.debug,
  29. 'INFO': self.logger.info,
  30. 'WARNING': self.logger.warning,
  31. 'ERROR': self.logger.error,
  32. 'CRITICAL': self.logger.critical}
  33. self._datavars: Optional[Datavars] = None
  34. # Словарь описаний команд.
  35. self.commands: Dict[str, Command] = self._get_commands_descriptions(
  36. config.commands_path)
  37. # Словарь CID и экземпляров команд, передаваемых воркерам.
  38. self.commands_runners: Dict[str, CommandRunner] = {}
  39. # Словарь WID и экземпляров процессов-воркеров, передаваемых воркерам.
  40. # TODO добавить менеджера воркеров.
  41. # self.workers: Dict[int, Worker] = {}
  42. @property
  43. def datavars(self):
  44. if self._datavars is not None:
  45. return self._datavars
  46. else:
  47. self._datavars = self._load_datavars(self._variables_path,
  48. self.logger)
  49. return self._datavars
  50. def _load_datavars(vars_path: str, logger: Logger) -> Datavars:
  51. return Datavars(variables_path=vars_path,
  52. logger=logger)
  53. def _get_commands_descriptions(self, commands_path: str
  54. ) -> Dict[str, Command]:
  55. '''Метод для получения совокупности описаний команд.'''
  56. output = {}
  57. package = ".".join(commands_path.split("/"))
  58. for entry in os.scandir(commands_path):
  59. if (not entry.name.endswith('.py')
  60. or entry.name in {"commands.py", "__init__.py"}):
  61. continue
  62. module_name = entry.name[:-3]
  63. try:
  64. module = importlib.import_module("{}.{}".format(package,
  65. module_name))
  66. for obj_name in dir(module):
  67. obj = module.__getattribute__(obj_name)
  68. if isinstance(obj, Command):
  69. output[obj.id] = obj
  70. except Exception:
  71. continue
  72. return output
  73. def make_command(self, command_id: str, ) -> int:
  74. '''Метод для создания команды по ее описанию.'''
  75. # command_description = self.commands[command_id]
  76. # def _get_worker_object(self, wid: Optional[int] = None) -> Worker:
  77. # '''Метод для получения воркера для команды.'''
  78. # if wid is not None:
  79. # worker = Worker(wid, self._event_loop, self._datavars)
  80. # self._workers[wid] = worker
  81. # elif not self._workers:
  82. # worker = Worker(0, self._event_loop, self._datavars)
  83. # self._workers[0] = worker
  84. # else:
  85. # wid = max(self._workers.keys()) + 1
  86. # worker = Worker(wid, self._event_loop, self._datavars)
  87. # self._workers[wid] = worker
  88. # return worker
  89. class Commands:
  90. '''Предварительная реализация контейнера описаний команд.'''
  91. def __init__(self, *commands: List[Command]):
  92. self._by_id: dict = dict()
  93. self._by_command: dict = dict()
  94. for command in commands:
  95. self._by_id[command.id] = command
  96. self._by_command[command.command] = command
  97. def get_commands(self, base_url: str) -> Dict[str, dict]:
  98. response = dict()
  99. for command in self._by_id.values():
  100. data = ResponseStructure(base_url)
  101. data.add_data(id=command.id,
  102. title=command.title,
  103. category=command.category,
  104. command=command.command)
  105. data.add_link("self", f"/commands/{command.id}?by_id=true")
  106. data.add_link("parameters", f"/commands/{command.id}/parameters")
  107. data.add_link("configure", f"/configs/{command.id}")
  108. response[command.id] = data.get_dict()
  109. return response
  110. def get_by_id(self, command_id: str, base_url: str) -> Dict[str, Dict]:
  111. if command_id in self._by_id:
  112. command = self._by_id[command_id]
  113. data = ResponseStructure(base_url)
  114. data.add_data(id=command.id,
  115. title=command.title,
  116. category=command.category,
  117. command=command.command)
  118. data.add_link("self", f"/commands/{command.id}?by_id=1")
  119. data.add_link("parameters", f"/commands/{command.id}/parameters")
  120. data.add_link("configure", f"/configs/{command.id}")
  121. parameters_data = self._get_parameters_data(command.parameters,
  122. command.id, base_url)
  123. data.embed("parameters", parameters_data)
  124. return data.get_dict()
  125. return None
  126. def find_command(self, console_command: str,
  127. base_url: str) -> Union[dict, None]:
  128. if console_command in self._by_command:
  129. command = self._by_command[console_command]
  130. data = ResponseStructure(base_url)
  131. data.add_data(id=command.id,
  132. title=command.title,
  133. category=command.category,
  134. command=command.command)
  135. data.add_link("self", f"/commands/{command.id}?by_id=1")
  136. data.add_link("parameters", f"/commands/{command.id}/parameters")
  137. data.add_link("configure", f"/configs/{command.id}")
  138. parameters_data = self._get_parameters_data(command.parameters,
  139. command.id, base_url)
  140. data.embed("parameters", parameters_data)
  141. return data.get_dict()
  142. else:
  143. return None
  144. def get_parameters(self, command_id: str, base_url: str
  145. ) -> Union[dict, None]:
  146. if command_id in self._by_id:
  147. command = self._by_id[command_id]
  148. parameters_data = self._get_parameters_data(command.parameters,
  149. command.id, base_url)
  150. return parameters_data
  151. else:
  152. return None
  153. def get_command_object(self, command_id: str) -> Command:
  154. if command_id in self._by_id:
  155. return self._by_id[command_id]
  156. return None
  157. def _get_parameters_data(self, parameters: list, command_id: str,
  158. base_url: str) -> List[dict]:
  159. data = ResponseStructure(base_url)
  160. data.add_data(parameters)
  161. data.add_link("self", f"/commands/{command_id}/parameters")
  162. return data.get_dict()