312 lines
8.8 KiB
Python
312 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Callable, Self, Sequence, Optional, ClassVar
|
|
from re import Pattern as REPattern
|
|
from threading import Thread
|
|
|
|
class RE:
|
|
|
|
KEY:ClassVar[REPattern]
|
|
STRING_VARIABLES:ClassVar[REPattern]
|
|
SLASHES:ClassVar[REPattern]
|
|
NEW_LINES:ClassVar[REPattern]
|
|
ROUTE:ClassVar[REPattern]
|
|
TO_REGULAR_EXPRESSION:ClassVar[REPattern]
|
|
ROUTE_KEY:ClassVar[REPattern]
|
|
EXCEPTION:ClassVar[REPattern]
|
|
NEW_LINE:ClassVar[REPattern]
|
|
HTTP_VARIABLE:ClassVar[REPattern]
|
|
PARENT_PATH:ClassVar[REPattern]
|
|
DOUBLE_NEW_LINE:ClassVar[REPattern]
|
|
HTTP_HEADER_PARAMETER:ClassVar[REPattern]
|
|
HTTP_REQUEST:ClassVar[REPattern]
|
|
|
|
class Check(ABC):
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_string(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_binary(item:Any|None) -> bool:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def is_key(cls:type[Self], item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_array(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_dictionary(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_boolean(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_integer(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_function(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_regular_expression(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_json_string(item:Any|None, full:bool = False) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_json(item:Any|None) -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_json_data(item:Any|None, full:bool = False) -> bool:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def is_mark_key(cls:type[Self], key:str, marks:str|Sequence[str] = "AnP") -> bool:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def is_file(path:str) -> bool:pass
|
|
|
|
class Common(ABC):
|
|
|
|
ROOT_PATH:ClassVar[str]
|
|
SLASH:ClassVar[str]
|
|
ROOTS_PATH:ClassVar[list[str]]
|
|
SPECIAL_REGULAR_EXPRESSION_CHARACTERS:ClassVar[dict[str, str]]
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def get_keys(cls:type[Self], *items:list[Any|None]) -> list[str]:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def get_texts(cls:type[Self], *items:list[Any|None]) -> list[str]:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def get_dictionaries(cls:type[Self], *items:list[Any|None]) -> list[dict[str, Any|None]]:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def get_dictionary(cls:type[Self], inputs:Any|None, overwrite:bool = False) -> dict[str, Any|None]:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def get_value(cls:type[Self],
|
|
keys:str|Sequence[str],
|
|
inputs:dict[str, Any|None]|Sequence[Any|None],
|
|
default:Optional[Any|None] = None
|
|
) -> Any|None:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def string_variables(cls:type[Self],
|
|
string:str,
|
|
inputs:dict[str, Any|None]|Sequence[Any|None],
|
|
default:Optional[str] = None
|
|
) -> str:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def fix_path(cls:type[Self], path:str) -> str:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def get_absolute_path(cls:type[Self], path:str) -> str|None:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def load_file(cls:type[Self], path:str, mode:str = "r") -> str|bytes|None:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def load_json(cls:type[Self],
|
|
data:str|dict[str, Any|None]|Sequence[Any|None],
|
|
only_dictionaries:bool = True
|
|
) -> list[dict[str, Any|None]|Sequence[Any|None]]:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def get_action_data(i:int = 0) -> dict[str, str|int]:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def to_regular_expression(cls:type[Self], string:str) -> str:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def get_mime_from_path(path:str) -> str|None:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def json_encode(data:dict[str, Any|None]|Sequence[Any|None]) -> str|None:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def json_decode(data:str|bytes) -> dict[str, Any|None]|Sequence[Any|None]|None:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def base64_encode(data:bytes) -> str|None:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def base64_decode(data:str) -> bytes|None:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def data_encode(cls:type[Self], data:Any|None) -> str|None:pass
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def data_decode(cls:type[Self], data:str) -> Any|None:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def execute(callback:Callable[..., Any|None], *arguments:Any|None) -> Any|None:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def unique(items:str|Sequence[Any|None]) -> str|list[Any|None]:pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def random(items:str|Sequence[Any|None]) -> Any|None:pass
|
|
|
|
class AIResponseModel(ABC):
|
|
|
|
def __init__(self:Self, conversation:str) -> None:
|
|
self.conversation:str
|
|
self.start:float
|
|
self.model:str
|
|
self.response:str
|
|
self.total:str
|
|
self.context:str|list[str]
|
|
self.done:bool
|
|
self.end:float|None
|
|
self.chunks:int
|
|
self.ok:bool
|
|
self.http_code:int
|
|
self.http_message:str
|
|
|
|
@abstractmethod
|
|
def update(self:Self, data:dict[str, Any|None]) -> None:pass
|
|
|
|
class CommandModel(ABC):
|
|
|
|
def __init__(self:Self,
|
|
names:str|Sequence[str],
|
|
callback:Callable[[dict[str, Any|None], Optional[list[Any|None]]], None]
|
|
) -> None:
|
|
self.names:list[str]
|
|
self.callback:Callable[[dict[str, Any|None], Optional[list[Any|None]]], None]
|
|
|
|
@abstractmethod
|
|
def update(self:Self,
|
|
names:str|Sequence[str],
|
|
callback:Optional[Callable[[dict[str, Any|None], Optional[list[Any|None]]], None]] = None,
|
|
overwrite:bool = False
|
|
) -> None:pass
|
|
|
|
class PseudoLoRAModel(ABC):
|
|
|
|
def __init__(self:Self,
|
|
title:str,
|
|
content:str|Sequence[Any|None],
|
|
keys:Optional[str|Sequence[str]] = None,
|
|
cacheable:bool = False
|
|
) -> None:
|
|
self.title:str
|
|
self.path:str|None
|
|
self.memory:int
|
|
self.cache:str
|
|
self.i:int
|
|
self.keys:list[str]
|
|
self.cacheable:bool
|
|
self.nested:list[PseudoLoRAModel]
|
|
|
|
@abstractmethod
|
|
def clean_cache(self:Self) -> None:pass
|
|
|
|
class QueueItemModel(ABC):
|
|
def __init__(self:Self, i:int, callback:Callable[[Callable[[], None]], None], *arguments:list[Any|None]) -> None:
|
|
self.i:int
|
|
self.callback:Callable[[Callable[[], None]], None]
|
|
self.arguments:list[Any|None]
|
|
self.executing:bool
|
|
self.thread:Thread|None
|
|
|
|
class QueuesModel(ABC):
|
|
|
|
def __init__(self:Self, key:str, inputs:Optional[int|dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
|
|
self.key:str
|
|
self.items:dict[int, QueueItemModel]
|
|
self.i:int
|
|
self.p:int
|
|
self.maximum:int
|
|
self.current:int
|
|
|
|
@abstractmethod
|
|
def add(self:Self, callback:Callable[[Callable[[], None]], None], *arguments:list[Any|None]) -> int|None:pass
|
|
|
|
@abstractmethod
|
|
def cancel(self:Self, i:int) -> bool:pass
|
|
|
|
@abstractmethod
|
|
def cancel_all(self:Self) -> None:pass
|
|
|
|
@abstractmethod
|
|
def next(self:Self) -> None:pass
|
|
|
|
class RequestModel(ABC):
|
|
|
|
def __init__(self:Self, session:Optional[SessionModel] = None) -> None:
|
|
self.post_variables:dict[str, Any|None]
|
|
self.get_variables:dict[str, Any|None]
|
|
self.url_variables:dict[str, Any|None]
|
|
self.hash_variables:dict[str, Any|None]
|
|
self.cookies:dict[str, Any|None]
|
|
self.variables:dict[str, Any|None]
|
|
self.request_headers:dict[str, Any|None]
|
|
self.method:str|None
|
|
self.route:RouteAbstract|None
|
|
self.response:str|bytes|None
|
|
self.response_mime:str|None
|
|
self.response_charset:str|None
|
|
self.response_code:int
|
|
self.response_headers:dict[str, Any|None]
|
|
self.callback:Callable[[RequestModel, str|bytes|None], None]|None
|
|
self.data:Any|None
|
|
self.session:SessionModel|None
|
|
self.protocol:str|None
|
|
self.protocol_version:str|None
|
|
self.last_modified:str|None
|
|
|
|
@abstractmethod
|
|
def get(self:Self, key:str|Sequence[str], default:Optional[Any] = None) -> Any|None:pass
|
|
|
|
@abstractmethod
|
|
def set_variables(self:Self, inputs:dict[str, Any|None], on:Optional[str] = None) -> None:pass
|
|
|
|
@abstractmethod
|
|
def set_response(self:Self, data:Any|None) -> None:pass
|
|
|
|
@abstractmethod
|
|
def set_request_headers(self:Self, inputs:Any|None) -> None:pass
|
|
|
|
@abstractmethod
|
|
def set_response_headers(self:Self, inputs:Any|None) -> None:pass |