diff --git a/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py new file mode 100644 index 0000000..4a45ef1 --- /dev/null +++ b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py @@ -0,0 +1,110 @@ +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