Удалить solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py

solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py

@@ -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)