#!/usr/bin/env python3 # -*- coding: utf-8 -*- from abc import ABC, abstractmethod from typing import Any, Self, Optional, Sequence from re import Match as REMath import datetime from Interfaces.Application.AnPInterface import AnPInterface from Utils.Common import Common class HTTPServersAbstract(ABC): DEFAULT_PORT:int = 8000 DEFAULT_HOST:str = "" def __init__(self:Self, anp:AnPInterface, key:str, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None: self.anp:AnPInterface = anp self._inputs:dict[str, Any|None] = Common.get_dictionary(inputs) self.host:str = self.DEFAULT_HOST self.port:int = self.DEFAULT_PORT self._print_data:dict[str, Any|None] = { "port": self.port, "host": self.host } self.http_messages:dict[int, str] = {int(code): message for code, message in anp.settings.get(( "default_http_messages", "http_messages" ), inputs).items()} self.protocol:str = anp.settings.get(("http_protocol", "protocol"), inputs, "HTTP") self.protocol_version:str = anp.settings.get(("http_protocol_version", "protocol_version"), inputs, "1.1") self.accept_range:str = anp.settings.get(("http_accept_range", "accept_range"), inputs, "bytes") self.access_control_max_age:int = anp.settings.get(("http_access_control_max_age", "access_control_max_age"), inputs, 84600) self.keep_alive_maximum:int = anp.settings.get(("http_keep_alive_maximum", "keep_alive_maximum"), inputs, 100) self.keep_alive_timeout:int = anp.settings.get(("http_keep_alive_timeout", "keep_alive_timeout"), inputs, 5) self.cors:str = anp.settings.get(("http_cors", "cors"), inputs, "*") self.mime:str = anp.settings.get(("default_http_mime", "http_mime", "mime"), inputs, "text/html") self.charset:str = anp.settings.get(("default_http_charset", "http_charset", "charset"), inputs, "UTF-8") self._session_timeout:int = anp.settings.get(("sessions_timeout", "timeout"), inputs, 3600) self.key:str = key self.update() @staticmethod def format_datetime(datetime:datetime.datetime) -> str: return datetime.strftime("%a, %d %b %Y %H:%M:%S GMT") def get_http_message(self:Self, code:int) -> str: return self.http_messages.get(code, "Unknown Status Code") def __update_print_data(self:Self) -> None: self._print_data["port"] = self.port self._print_data["host"] = self.host @abstractmethod def start(self:Self) -> None:pass @abstractmethod def close(self:Self) -> None:pass def update(self:Self) -> None: self.close() self.port = self.anp.settings.get(("http_server_port", "http_port", "port"), self._inputs, self.DEFAULT_PORT) self.host = self.anp.settings.get(("http_server_host", "http_host", "host"), self._inputs, self.DEFAULT_HOST) self.__update_print_data() self.anp.settings.get(("http_server_autostart", "http_autostart", "autostart")) and self.start() def reset(self:Self) -> None: self.port = self.DEFAULT_PORT self.host = self.DEFAULT_HOST self.__update_print_data() self.update() def load_cookies(self:Self, cookies:str|None) -> dict[str, Any|None]: if not cookies: return {} cookie:str results:dict[str, Any|None] = {} for cookie in cookies.split(";"): if "=" in cookie: key:str value:str key, value = cookie.split("=", 1) results[key.strip()] = value.strip() return results @staticmethod def get_variables_from(source:str|None) -> dict[str, Any|None]: if not source: return {} json:dict[str, Any|None] = Common.data_decode(source) if json is not None: return Common.get_dictionary(json) json = {} def callback(matches:REMath) -> str: json[matches.group(1)] = matches.group(2) return matches.group(0) Common.replace(self.HTTP_VARIABLE, source, callback) return json