forked from Shifty/pyserveX
156 lines
5.4 KiB
Python
156 lines
5.4 KiB
Python
import yaml
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List, Optional
|
|
from dataclasses import dataclass, field
|
|
import logging
|
|
from .logging_utils import setup_logging
|
|
|
|
|
|
@dataclass
|
|
class HttpConfig:
|
|
static_dir: str = "./static"
|
|
templates_dir: str = "./templates"
|
|
|
|
|
|
@dataclass
|
|
class ServerConfig:
|
|
host: str = "0.0.0.0"
|
|
port: int = 8080
|
|
backlog: int = 5
|
|
default_root: bool = False
|
|
redirect_instructions: Dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class SSLConfig:
|
|
enabled: bool = False
|
|
cert_file: str = "./ssl/cert.pem"
|
|
key_file: str = "./ssl/key.pem"
|
|
|
|
|
|
@dataclass
|
|
class LoggingConfig:
|
|
level: str = "INFO"
|
|
console_output: bool = True
|
|
log_file: str = "./logs/pyserve.log"
|
|
|
|
|
|
@dataclass
|
|
class RoutingExtensionConfig:
|
|
regex_locations: Dict[str, Dict[str, Any]] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class ExtensionConfig:
|
|
type: str
|
|
config: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
http: HttpConfig = field(default_factory=HttpConfig)
|
|
server: ServerConfig = field(default_factory=ServerConfig)
|
|
ssl: SSLConfig = field(default_factory=SSLConfig)
|
|
logging: LoggingConfig = field(default_factory=LoggingConfig)
|
|
extensions: List[ExtensionConfig] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_yaml(cls, file_path: str) -> "Config":
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
return cls._from_dict(data)
|
|
except FileNotFoundError:
|
|
logging.warning(f"Конфигурационный файл {file_path} не найден. Используются значения по умолчанию.")
|
|
return cls()
|
|
except yaml.YAMLError as e:
|
|
logging.error(f"Ошибка парсинга YAML файла {file_path}: {e}")
|
|
raise
|
|
|
|
@classmethod
|
|
def _from_dict(cls, data: Dict[str, Any]) -> "Config":
|
|
config = cls()
|
|
|
|
if 'http' in data:
|
|
http_data = data['http']
|
|
config.http = HttpConfig(
|
|
static_dir=http_data.get('static_dir', config.http.static_dir),
|
|
templates_dir=http_data.get('templates_dir', config.http.templates_dir)
|
|
)
|
|
|
|
if 'server' in data:
|
|
server_data = data['server']
|
|
config.server = ServerConfig(
|
|
host=server_data.get('host', config.server.host),
|
|
port=server_data.get('port', config.server.port),
|
|
backlog=server_data.get('backlog', config.server.backlog),
|
|
default_root=server_data.get('default_root', config.server.default_root),
|
|
redirect_instructions=server_data.get('redirect_instructions', {})
|
|
)
|
|
|
|
if 'ssl' in data:
|
|
ssl_data = data['ssl']
|
|
config.ssl = SSLConfig(
|
|
enabled=ssl_data.get('enabled', config.ssl.enabled),
|
|
cert_file=ssl_data.get('cert_file', config.ssl.cert_file),
|
|
key_file=ssl_data.get('key_file', config.ssl.key_file)
|
|
)
|
|
|
|
if 'logging' in data:
|
|
log_data = data['logging']
|
|
config.logging = LoggingConfig(
|
|
level=log_data.get('level', config.logging.level),
|
|
console_output=log_data.get('console_output', config.logging.console_output),
|
|
log_file=log_data.get('log_file', config.logging.log_file)
|
|
)
|
|
|
|
if 'extensions' in data:
|
|
for ext_data in data['extensions']:
|
|
extension = ExtensionConfig(
|
|
type=ext_data.get('type', ''),
|
|
config=ext_data.get('config', {})
|
|
)
|
|
config.extensions.append(extension)
|
|
|
|
return config
|
|
|
|
def validate(self) -> bool:
|
|
errors = []
|
|
|
|
if not os.path.exists(self.http.static_dir):
|
|
errors.append(f"Статическая директория не существует: {self.http.static_dir}")
|
|
|
|
if self.ssl.enabled:
|
|
if not os.path.exists(self.ssl.cert_file):
|
|
errors.append(f"SSL сертификат не найден: {self.ssl.cert_file}")
|
|
if not os.path.exists(self.ssl.key_file):
|
|
errors.append(f"SSL ключ не найден: {self.ssl.key_file}")
|
|
|
|
if not (1 <= self.server.port <= 65535):
|
|
errors.append(f"Некорректный порт: {self.server.port}")
|
|
|
|
log_dir = os.path.dirname(self.logging.log_file)
|
|
if log_dir and not os.path.exists(log_dir):
|
|
try:
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
except OSError as e:
|
|
errors.append(f"Невозможно создать директорию для логов: {e}")
|
|
|
|
if errors:
|
|
for error in errors:
|
|
logging.error(f"Ошибка конфигурации: {error}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def setup_logging(self) -> None:
|
|
"""Настройка системы логирования через кастомный менеджер"""
|
|
config_dict = {
|
|
'level': self.logging.level,
|
|
'console_output': self.logging.console_output,
|
|
'log_file': self.logging.log_file
|
|
}
|
|
setup_logging(config_dict)
|