diff --git a/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py deleted file mode 100644 index 4a45ef1..0000000 --- a/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py +++ /dev/null @@ -1,110 +0,0 @@ -import json -from typing import Any, Dict, Callable, List - -class MiddlewareError(Exception): - """Исключение при работе middleware.""" - - -class HumanInTheLoop: - """ - Класс реализует цепочку обработки данных с возможностью вмешательства человека. - """ - - def __init__(self) -> None: - self._middleware: List[Callable[[Dict[str, Any]], Dict[str, Any]]] = [] - self._handlers: Dict[str, Callable[[Dict[str, Any]], None]] = {} - - def add_middleware(self, func: Callable[[Dict[str, Any]], Dict[str, Any]]) -> None: - """ - Добавляет функцию middleware в цепочку. - """ - if not callable(func): - raise MiddlewareError("Middleware должен быть вызываемым.") - self._middleware.append(func) - - def register_handler(self, name: str, handler: Callable[[Dict[str, Any]], None]) -> None: - """ - Регистрирует обработчик с именем `name`. - """ - if not callable(handler): - raise MiddlewareError("Handler должен быть вызываемым.") - self._handlers[name] = handler - - def _process(self, data: Dict[str, Any]) -> Dict[str, Any]: - """ - Последовательно применяет все middleware к данным. - """ - for mw in self._middleware: - try: - data = mw(data) - if not isinstance(data, dict): - raise MiddlewareError("Middleware должен вернуть словарь.") - except Exception as exc: - raise MiddlewareError(f"Ошибка в middleware {mw.__name__}: {exc}") from exc - return data - - def run(self, raw_input: str) -> None: - """ - Точка входа. Принимает строку JSON, преобразует её, - запускает цепочку middleware и вызывает соответствующий обработчик. - """ - try: - payload = json.loads(raw_input) - except json.JSONDecodeError as exc: - raise MiddlewareError("Неверный формат JSON") from exc - - processed = self._process(payload) - - handler_name = processed.get("handler") - if not handler_name or handler_name not in self._handlers: - raise MiddlewareError(f"Неизвестный обработчик: {handler_name}") - - try: - self._handlers[handler_name](processed) - except Exception as exc: - raise MiddlewareError(f"Ошибка в обработчике {handler_name}: {exc}") from exc - -# ----------------- Пример middleware и обработчиков ----------------- - -def validate_mw(data: Dict[str, Any]) -> Dict[str, Any]: - """Проверяет наличие обязательных полей.""" - if "payload" not in data: - raise MiddlewareError("Отсутствует поле 'payload'") - return data - -def enrich_mw(data: Dict[str, Any]) -> Dict[str, Any]: - """Добавляет метку времени.""" - import datetime - data["timestamp"] = datetime.datetime.utcnow().isoformat() - return data - -def echo_handler(data: Dict[str, Any]) -> None: - print("Echo:", data.get("payload")) - -def sum_handler(data: Dict[str, Any]) -> None: - numbers = data.get("payload", []) - if not isinstance(numbers, list): - raise MiddlewareError("'payload' должен быть списком") - result = sum(numbers) - print(f"Sum: {result}") - -# ----------------- Инициализация ----------------- -if __name__ == "__main__": - hltl = HumanInTheLoop() - hltl.add_middleware(validate_mw) - hltl.add_middleware(enrich_mw) - - hltl.register_handler("echo", echo_handler) - hltl.register_handler("sum", sum_handler) - - # Пример входных данных - test_inputs = [ - json.dumps({"handler": "echo", "payload": "Hello, world!"}), - json.dumps({"handler": "sum", "payload": [1, 2, 3, 4]}), - ] - - for inp in test_inputs: - try: - hltl.run(inp) - except MiddlewareError as e: - print("Ошибка:", e) \ No newline at end of file