From 5e217c7cce3576d17f613076b1c904b179155f74 Mon Sep 17 00:00:00 2001 From: Maksim Date: Mon, 1 Sep 2025 19:19:28 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BF=D0=BE=D1=80=D1=82=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python_parser/core/ports.py | 89 ++++++++++++++++++++++++++++++++----- 1 file changed, 77 insertions(+), 12 deletions(-) diff --git a/python_parser/core/ports.py b/python_parser/core/ports.py index 4519374..c3a5c67 100644 --- a/python_parser/core/ports.py +++ b/python_parser/core/ports.py @@ -2,28 +2,93 @@ Порты (интерфейсы) для hexagonal architecture """ from abc import ABC, abstractmethod -from typing import Optional +from typing import Optional, Dict, List, Any, Callable import pandas as pd class ParserPort(ABC): - """Интерфейс для парсеров""" + """Интерфейс для парсеров с поддержкой множественных геттеров""" + + def __init__(self): + """Инициализация с пустым словарем геттеров""" + self.getters: Dict[str, Dict[str, Any]] = {} + self._register_default_getters() + + def _register_default_getters(self): + """Регистрация геттеров по умолчанию - переопределяется в наследниках""" + pass + + def register_getter(self, name: str, method: Callable, required_params: List[str], + optional_params: List[str] = None, description: str = ""): + """ + Регистрация нового геттера + + Args: + name: Имя геттера + method: Метод для выполнения + required_params: Список обязательных параметров + optional_params: Список необязательных параметров + description: Описание геттера + """ + if optional_params is None: + optional_params = [] + + self.getters[name] = { + "method": method, + "required_params": required_params, + "optional_params": optional_params, + "description": description + } + + def get_available_getters(self) -> Dict[str, Dict[str, Any]]: + """Получение списка доступных геттеров с их описанием""" + return { + name: { + "required_params": info["required_params"], + "optional_params": info["optional_params"], + "description": info["description"] + } + for name, info in self.getters.items() + } + + # Добавить схему + def get_value(self, getter_name: str, params: Dict[str, Any]): + """ + Получение значения через указанный геттер + + Args: + getter_name: Имя геттера + params: Параметры для геттера + + Returns: + Результат выполнения геттера + + Raises: + ValueError: Если геттер не найден или параметры неверны + """ + if getter_name not in self.getters: + available = list(self.getters.keys()) + raise ValueError(f"Геттер '{getter_name}' не найден. Доступные: {available}") + + getter_info = self.getters[getter_name] + required = getter_info["required_params"] + + # Проверка обязательных параметров + missing = [p for p in required if p not in params] + if missing: + raise ValueError(f"Отсутствуют обязательные параметры для геттера '{getter_name}': {missing}") + + # Вызов метода геттера + try: + return getter_info["method"](params) + except Exception as e: + raise ValueError(f"Ошибка выполнения геттера '{getter_name}': {str(e)}") @abstractmethod def parse(self, file_path: str, params: dict) -> pd.DataFrame: """Парсинг файла и возврат DataFrame""" pass - @abstractmethod - def get_value(self, df: pd.DataFrame, params: dict): - """Получение значения из DataFrame по параметрам""" - pass - - # @abstractmethod - # def get_schema(self) -> dict: - # """Возвращает схему входных параметров для парсера""" - # pass - class StoragePort(ABC): """Интерфейс для хранилища данных"""