Maks/python_parser

Fork 0

Files

Maksim b5c460bb6f monitoring_tar полностью функционален

2025-09-04 12:57:28 +03:00

36 KiB

Raw Permalink Blame History

📚 Руководство по разработке парсеров

Полное руководство по созданию новых парсеров для системы NIN Excel Parsers API.

📋 Содержание

Архитектура системы
Структура проекта
Создание нового парсера
Регистрация парсера
Создание API эндпоинтов
Интеграция с Streamlit
Тестирование
Лучшие практики
Примеры

🏗️ Архитектура системы

Hexagonal Architecture

Система построена на принципах Hexagonal Architecture (Ports & Adapters):

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Streamlit UI  │    │   FastAPI       │    │   MinIO Storage │
│   (Adapter)     │◄──►│   (Application) │◄──►│   (Adapter)     │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │   ParserPort    │
                    │   (Core)        │
                    └─────────────────┘
                              ▲
                              │
                    ┌─────────────────┐
                    │   Parser        │
                    │   (Adapter)     │
                    └─────────────────┘

Ключевые компоненты

ParserPort - базовый класс для всех парсеров
ReportService - сервис для управления отчетами
MinIOStorageAdapter - адаптер для хранения данных
FastAPI - веб-фреймворк для API
Streamlit - веб-интерфейс

📁 Структура проекта

python_parser/
├── adapters/
│   ├── parsers/           # Парсеры (адаптеры)
│   │   ├── __init__.py
│   │   ├── monitoring_fuel.py
│   │   ├── monitoring_tar.py
│   │   ├── svodka_ca.py
│   │   ├── svodka_pm.py
│   │   ├── svodka_repair_ca.py
│   │   └── statuses_repair_ca.py
│   ├── pconfig.py         # Конфигурация парсеров
│   └── storage.py         # Адаптер хранилища
├── app/
│   ├── main.py           # FastAPI приложение
│   └── schemas/          # Pydantic схемы
│       ├── monitoring_fuel.py
│       ├── monitoring_tar.py
│       ├── svodka_ca.py
│       ├── svodka_pm.py
│       ├── svodka_repair_ca.py
│       └── statuses_repair_ca.py
├── core/
│   ├── models.py         # Модели данных
│   ├── ports.py          # Базовые порты
│   ├── schema_utils.py   # Утилиты схем
│   └── services.py       # Сервисы
└── requirements.txt

streamlit_app/
├── streamlit_app.py      # Streamlit интерфейс
└── requirements.txt

🔧 Создание нового парсера

1. Создание Pydantic схем

Создайте файл схемы в python_parser/app/schemas/your_parser.py:

from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from enum import Enum

class YourParserMode(str, Enum):
    """Режимы работы парсера"""
    MODE1 = "mode1"
    MODE2 = "mode2"

class YourParserRequest(BaseModel):
    """Схема запроса для основного геттера"""
    param1: Optional[str] = Field(
        None,
        description="Описание параметра 1",
        example="example_value"
    )
    param2: Optional[List[str]] = Field(
        None,
        description="Описание параметра 2",
        example=["value1", "value2"]
    )
    mode: Optional[YourParserMode] = Field(
        None,
        description="Режим работы парсера",
        example="mode1"
    )

    class Config:
        json_schema_extra = {
            "example": {
                "param1": "example_value",
                "param2": ["value1", "value2"],
                "mode": "mode1"
            }
        }

class YourParserFullRequest(BaseModel):
    """Схема запроса для получения всех данных"""
    # Пустая схема - возвращает все данные без фильтрации
    pass

    class Config:
        json_schema_extra = {
            "example": {}
        }

2. Создание парсера

Создайте файл парсера в python_parser/adapters/parsers/your_parser.py:

import pandas as pd
import os
import zipfile
import tempfile
from typing import Dict, Any, Optional
from core.ports import ParserPort
from adapters.pconfig import find_header_row, data_to_json
from app.schemas.your_parser import YourParserRequest, YourParserFullRequest


class YourParser(ParserPort):
    """Парсер для вашего типа данных"""
    
    name = "your_parser"  # Уникальное имя парсера
    
    def __init__(self):
        super().__init__()
        # Регистрируем геттеры
        self.register_getter("get_data", YourParserRequest, self._get_data_wrapper)
        self.register_getter("get_full_data", YourParserFullRequest, self._get_full_data_wrapper)
        
        # Данные парсера
        self.data_dict = {}
    
    def parse(self, file_path: str, params: dict) -> Dict[str, Any]:
        """Основной метод парсинга"""
        print(f"🔍 DEBUG: YourParser.parse вызван с файлом: {file_path}")
        
        try:
            # Проверяем тип файла (пример для ZIP-only парсера)
            if not file_path.endswith('.zip'):
                raise ValueError(f"Неподдерживаемый тип файла: {file_path}. Ожидается только ZIP архив.")
            
            # Обрабатываем ZIP архив
            result = self._parse_zip_archive(file_path)
            
            # Сохраняем результат
            self.data_dict = result
            print(f"✅ Парсинг завершен. Получено {len(result)} записей")
            return result
            
        except Exception as e:
            print(f"❌ Ошибка при парсинге: {e}")
            raise
    
    def _parse_zip_archive(self, zip_path: str) -> Dict[str, Any]:
        """Парсинг ZIP архива"""
        print(f"📦 Обработка ZIP архива: {zip_path}")
        
        with tempfile.TemporaryDirectory() as temp_dir:
            with zipfile.ZipFile(zip_path, 'r') as zip_ref:
                zip_ref.extractall(temp_dir)
            
            # Ищем нужные файлы
            target_files = []
            for root, dirs, files in os.walk(temp_dir):
                for file in files:
                    if file.endswith(('.xlsx', '.xls')):
                        target_files.append(os.path.join(root, file))
            
            if not target_files:
                raise ValueError("В архиве не найдены поддерживаемые файлы")
            
            # Парсим все найденные файлы
            all_data = {}
            for file_path in target_files:
                file_data = self._parse_single_file(file_path)
                # Объединяем данные
                all_data.update(file_data)
            
            return all_data
    
    def _parse_single_file(self, file_path: str) -> Dict[str, Any]:
        """Парсинг одного файла"""
        print(f"📁 Обработка файла: {file_path}")
        
        try:
            # Читаем Excel файл
            excel_file = pd.ExcelFile(file_path)
            available_sheets = excel_file.sheet_names
            
            # Обрабатываем нужные листы
            result_data = {}
            for sheet_name in available_sheets:
                if self._should_process_sheet(sheet_name):
                    sheet_data = self._parse_sheet(file_path, sheet_name)
                    result_data.update(sheet_data)
            
            return result_data
            
        except Exception as e:
            print(f"❌ Ошибка при обработке файла {file_path}: {e}")
            return {}
    
    def _should_process_sheet(self, sheet_name: str) -> bool:
        """Определяет, нужно ли обрабатывать лист"""
        # Логика фильтрации листов
        return True  # Или ваша логика
    
    def _parse_sheet(self, file_path: str, sheet_name: str) -> Dict[str, Any]:
        """Парсинг конкретного листа"""
        try:
            # Находим заголовок
            header_num = find_header_row(file_path, sheet_name, search_value="1")
            if header_num is None:
                print(f"❌ Не найден заголовок в листе {sheet_name}")
                return {}
            
            # Читаем данные
            df = pd.read_excel(
                file_path,
                sheet_name=sheet_name,
                header=header_num,
                index_col=None
            )
            
            # Обрабатываем данные
            processed_data = self._process_dataframe(df, sheet_name)
            
            return processed_data
            
        except Exception as e:
            print(f"❌ Ошибка при обработке листа {sheet_name}: {e}")
            return {}
    
    def _process_dataframe(self, df: pd.DataFrame, sheet_name: str) -> Dict[str, Any]:
        """Обработка DataFrame"""
        # Ваша логика обработки данных
        return {"sheet_name": sheet_name, "data": df.to_dict('records')}
    
    def _get_data_wrapper(self, params: dict) -> Dict[str, Any]:
        """Обертка для основного геттера"""
        print(f"🔍 DEBUG: _get_data_wrapper вызван с параметрами: {params}")
        
        # Валидируем параметры
        validated_params = YourParserRequest(**params)
        
        # Получаем данные из парсера
        data_source = self._get_data_source()
        
        if not data_source:
            print("⚠️ Нет данных в парсере")
            return {}
        
        # Фильтруем данные по параметрам
        filtered_data = self._filter_data(data_source, validated_params)
        
        # Конвертируем в JSON
        try:
            result_json = data_to_json(filtered_data)
            return result_json
        except Exception as e:
            print(f"❌ Ошибка при конвертации данных в JSON: {e}")
            return {}
    
    def _get_full_data_wrapper(self, params: dict) -> Dict[str, Any]:
        """Обертка для геттера всех данных"""
        print(f"🔍 DEBUG: _get_full_data_wrapper вызван с параметрами: {params}")
        
        # Получаем данные из парсера
        data_source = self._get_data_source()
        
        if not data_source:
            print("⚠️ Нет данных в парсере")
            return {}
        
        # Конвертируем все данные в JSON
        try:
            result_json = data_to_json(data_source)
            return result_json
        except Exception as e:
            print(f"❌ Ошибка при конвертации данных в JSON: {e}")
            return {}
    
    def _get_data_source(self) -> Dict[str, Any]:
        """Получает источник данных"""
        if hasattr(self, 'df') and self.df is not None:
            # Данные загружены из MinIO
            if isinstance(self.df, dict):
                return self.df
            else:
                return {}
        elif hasattr(self, 'data_dict') and self.data_dict:
            # Данные из локального парсинга
            return self.data_dict
        else:
            return {}
    
    def _filter_data(self, data_source: Dict[str, Any], params: YourParserRequest) -> Dict[str, Any]:
        """Фильтрует данные по параметрам"""
        # Ваша логика фильтрации
        return data_source

📝 Регистрация парсера

1. Регистрация в init.py

Добавьте импорт в python_parser/adapters/parsers/__init__.py:

from .monitoring_fuel import MonitoringFuelParser
from .monitoring_tar import MonitoringTarParser
from .your_parser import YourParser  # Добавить эту строку
from .svodka_ca import SvodkaCAParser
from .svodka_pm import SvodkaPMParser
from .svodka_repair_ca import SvodkaRepairCAParser
from .statuses_repair_ca import StatusesRepairCAParser

__all__ = [
    'MonitoringFuelParser',
    'MonitoringTarParser',
    'YourParser',  # Добавить эту строку
    'SvodkaCAParser',
    'SvodkaPMParser',
    'SvodkaRepairCAParser',
    'StatusesRepairCAParser'
]

2. Регистрация в main.py

Добавьте импорт и регистрацию в python_parser/app/main.py:

# Импорты
from adapters.parsers import SvodkaPMParser, SvodkaCAParser, MonitoringFuelParser, MonitoringTarParser, YourParser, SvodkaRepairCAParser, StatusesRepairCAParser

from app.schemas.your_parser import YourParserRequest, YourParserFullRequest

# Регистрация парсера
PARSERS.update({
    'svodka_pm': SvodkaPMParser,
    'svodka_ca': SvodkaCAParser,
    'monitoring_fuel': MonitoringFuelParser,
    'monitoring_tar': MonitoringTarParser,
    'your_parser': YourParser,  # Добавить эту строку
    'svodka_repair_ca': SvodkaRepairCAParser,
    'statuses_repair_ca': StatusesRepairCAParser,
})

3. Регистрация в services.py

Добавьте логику выбора геттера в python_parser/core/services.py:

elif request.report_type == 'monitoring_tar':
    # Для monitoring_tar используем геттер get_tar_data
    getter_name = 'get_tar_data'
elif request.report_type == 'your_parser':  # Добавить эту секцию
    # Для your_parser используем геттер get_data
    getter_name = 'get_data'
elif request.report_type == 'monitoring_fuel':

🌐 Создание API эндпоинтов

Добавьте эндпоинты в python_parser/app/main.py:

# ====== YOUR PARSER ENDPOINTS ======

@app.post("/your_parser/upload", tags=[YourParser.name],
          summary="Загрузка отчета вашего типа")
async def upload_your_parser(
    file: UploadFile = File(...)
):
    """Загрузка и обработка отчета вашего типа

    ### Поддерживаемые форматы:
    - **ZIP архивы** с файлами (только ZIP)

    ### Структура данных:
    - Описание структуры данных
    - Какие данные извлекаются
    - Формат возвращаемых данных
    """
    report_service = get_report_service()

    try:
        # Проверяем тип файла - только ZIP архивы
        if not file.filename.endswith('.zip'):
            raise HTTPException(
                status_code=400,
                detail="Неподдерживаемый тип файла. Ожидается только ZIP архив (.zip)"
            )

        # Читаем содержимое файла
        file_content = await file.read()

        # Создаем запрос на загрузку
        upload_request = UploadRequest(
            report_type='your_parser',
            file_content=file_content,
            file_name=file.filename
        )

        # Загружаем отчет
        result = report_service.upload_report(upload_request)

        if result.success:
            return UploadResponse(
                success=True,
                message="Отчет успешно загружен и обработан",
                report_id=result.object_id,
                filename=file.filename
            ).model_dump()
        else:
            return UploadErrorResponse(
                success=False,
                message=result.message,
                error_code="ERR_UPLOAD",
                details=None
            ).model_dump()

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}")


@app.post("/your_parser/get_data", tags=[YourParser.name],
          summary="Получение данных из отчета")
async def get_your_parser_data(
    request_data: YourParserRequest
):
    """Получение данных из отчета

    ### Структура параметров:
    - `param1`: **Описание параметра 1** (опциональный)
    - `param2`: **Описание параметра 2** (опциональный)
    - `mode`: **Режим работы** (опциональный)

    ### Пример тела запроса:
    ```json
    {
        "param1": "example_value",
        "param2": ["value1", "value2"],
        "mode": "mode1"
    }
    ```
    """
    report_service = get_report_service()

    try:
        # Создаем запрос
        request_dict = request_data.model_dump()
        request = DataRequest(
            report_type='your_parser',
            get_params=request_dict
        )

        # Получаем данные
        result = report_service.get_data(request)

        if result.success:
            return {
                "success": True,
                "data": result.data
            }
        else:
            raise HTTPException(status_code=404, detail=result.message)

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}")


@app.post("/your_parser/get_full_data", tags=[YourParser.name],
          summary="Получение всех данных из отчета")
async def get_your_parser_full_data():
    """Получение всех данных из отчета без фильтрации

    ### Возвращает:
    - Все данные без фильтрации
    - Полная структура данных
    """
    report_service = get_report_service()

    try:
        # Создаем запрос без параметров
        request = DataRequest(
            report_type='your_parser',
            get_params={}
        )

        # Получаем данные через геттер get_full_data
        result = report_service.get_data(request, getter_name='get_full_data')

        if result.success:
            return {
                "success": True,
                "data": result.data
            }
        else:
            raise HTTPException(status_code=404, detail=result.message)

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}")

🎨 Интеграция с Streamlit

1. Добавление новой вкладки

В streamlit_app/streamlit_app.py:

# Обновите список вкладок
tab1, tab2, tab3, tab4, tab5, tab6, tab7 = st.tabs([
    "📊 Сводки ПМ", 
    "🏭 Сводки СА", 
    "⛽ Мониторинг топлива",
    "⚡ Мониторинг ТЭР",
    "🔧 Ремонт СА",
    "📋 Статусы ремонта СА",
    "🆕 Ваш парсер"  # Добавить новую вкладку
])

# Добавьте новую вкладку
with tab7:
    st.header("🆕 Ваш парсер")
    
    # Секция загрузки файлов
    st.subheader("📤 Загрузка файлов")
            uploaded_file = st.file_uploader(
            "Выберите ZIP архив для вашего парсера",
            type=['zip'],
            key="your_parser_upload"
        )
    
    if uploaded_file is not None:
        if st.button("📤 Загрузить файл", key="your_parser_upload_btn"):
            with st.spinner("Загружаем файл..."):
                file_data = uploaded_file.read()
                result, status_code = upload_file_to_api("/your_parser/upload", file_data, uploaded_file.name)
                
                if status_code == 200:
                    st.success("✅ Файл успешно загружен!")
                    st.json(result)
                else:
                    st.error(f"❌ Ошибка загрузки: {result}")
    
    # Секция получения данных
    st.subheader("📊 Получение данных")
    
    col1, col2 = st.columns(2)
    
    with col1:
        st.subheader("🔍 Фильтрованные данные")
        
        # Параметры запроса
        param1 = st.text_input("Параметр 1:", key="your_parser_param1")
        param2 = st.multiselect("Параметр 2:", ["value1", "value2", "value3"], key="your_parser_param2")
        mode = st.selectbox("Режим:", ["mode1", "mode2"], key="your_parser_mode")
        
        if st.button("📊 Получить данные", key="your_parser_get_data_btn"):
            with st.spinner("Получаем данные..."):
                request_data = {
                    "param1": param1 if param1 else None,
                    "param2": param2 if param2 else None,
                    "mode": mode if mode else None
                }
                result, status_code = make_api_request("/your_parser/get_data", request_data)
                
                if status_code == 200 and result.get("success"):
                    st.success("✅ Данные успешно получены!")
                    
                    # Показываем данные
                    data = result.get("data", {}).get("value", {})
                    if data:
                        st.subheader("📋 Результат:")
                        st.json(data)
                    else:
                        st.info("📋 Нет данных для отображения")
                else:
                    st.error(f"❌ Ошибка: {result.get('message', 'Неизвестная ошибка')}")
    
    with col2:
        st.subheader("📋 Все данные")
        
        if st.button("📊 Получить все данные", key="your_parser_get_full_data_btn"):
            with st.spinner("Получаем все данные..."):
                result, status_code = make_api_request("/your_parser/get_full_data", {})
                
                if status_code == 200 and result.get("success"):
                    st.success("✅ Все данные успешно получены!")
                    
                    # Показываем данные
                    data = result.get("data", {}).get("value", {})
                    if data:
                        st.subheader("📋 Результат:")
                        st.json(data)
                    else:
                        st.info("📋 Нет данных для отображения")
                else:
                    st.error(f"❌ Ошибка: {result.get('message', 'Неизвестная ошибка')}")

2. Обновление информации о проекте

# В секции "О проекте"
**Возможности:**
- 📊 Парсинг сводок ПМ (план и факт)
- 🏭 Парсинг сводок СА
- ⛽ Мониторинг топлива
- ⚡ Мониторинг ТЭР (Топливно-энергетические ресурсы)
- 🔧 Управление ремонтными работами СА
- 📋 Мониторинг статусов ремонта СА
- 🆕 Ваш новый парсер  # Добавить эту строку

🧪 Тестирование

1. Создание тестового скрипта

Создайте test_your_parser.py:

#!/usr/bin/env python3
"""
Тестовый скрипт для проверки парсера Your Parser
"""

import requests
import os

API_BASE_URL = "http://localhost:8000"

def test_upload_file():
    """Тест загрузки файла"""
    print("🔍 Тестируем загрузку файла...")
    
    # Используем тестовый файл
    file_path = "path/to/your/test/file.xlsx"
    
    if not os.path.exists(file_path):
        print(f"❌ Файл не найден: {file_path}")
        return False
    
    print(f"📁 Найден файл: {file_path}")
    
    with open(file_path, 'rb') as f:
        file_content = f.read()
    
    files = {
        'file': ('test_file.xlsx', file_content, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
    }
    
    try:
        response = requests.post(f"{API_BASE_URL}/your_parser/upload", files=files)
        print(f"📊 Статус ответа: {response.status_code}")
        print(f"📄 Ответ: {response.text}")
        
        if response.status_code == 200:
            result = response.json()
            if result.get("success"):
                print("✅ Загрузка успешна!")
                return True
            else:
                print("❌ Ошибка загрузки!")
                return False
        else:
            print("❌ Ошибка HTTP!")
            return False
            
    except Exception as e:
        print(f"❌ Исключение: {e}")
        return False

def test_get_data():
    """Тест получения данных"""
    print("\n🔍 Тестируем получение данных...")
    
    test_data = {
        "param1": "test_value",
        "param2": ["value1", "value2"],
        "mode": "mode1"
    }
    
    try:
        response = requests.post(f"{API_BASE_URL}/your_parser/get_data", json=test_data)
        print(f"📊 Статус ответа: {response.status_code}")
        
        if response.status_code == 200:
            result = response.json()
            if result.get("success"):
                print("✅ Получение данных успешно!")
                data = result.get("data", {}).get("value", {})
                print(f"📊 Получено данных: {len(data)} записей")
                return True
            else:
                print("❌ Ошибка получения данных!")
                print(f"📄 Ответ: {response.text}")
                return False
        else:
            print("❌ Ошибка HTTP!")
            print(f"📄 Ответ: {response.text}")
            return False
            
    except Exception as e:
        print(f"❌ Исключение: {e}")
        return False

def main():
    """Основная функция тестирования"""
    print("🚀 Тестирование парсера Your Parser")
    print("=" * 70)
    
    # Тест 1: Загрузка файла
    upload_success = test_upload_file()
    
    if upload_success:
        # Тест 2: Получение данных
        get_data_success = test_get_data()
        
        print("\n" + "=" * 70)
        print("📋 Результаты тестирования:")
        print(f"  ✅ Загрузка файла: {'ПРОЙДЕН' if upload_success else 'ПРОВАЛЕН'}")
        print(f"  ✅ Получение данных: {'ПРОЙДЕН' if get_data_success else 'ПРОВАЛЕН'}")
        
        if upload_success and get_data_success:
            print("\n🎉 Все тесты пройдены! Парсер работает корректно!")
        else:
            print("\n❌ Некоторые тесты провалены.")
    else:
        print("\n❌ Загрузка файла провалена. Пропускаем остальные тесты.")

if __name__ == "__main__":
    main()

2. Запуск тестов

# Запуск тестового скрипта
python test_your_parser.py

# Проверка через curl
curl -X POST "http://localhost:8000/your_parser/upload" \
  -H "accept: application/json" \
  -H "Content-Type: multipart/form-data" \
  -F "file=@test_file.xlsx"

curl -X POST "http://localhost:8000/your_parser/get_data" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{"param1": "test_value", "mode": "mode1"}'

📋 Лучшие практики

1. Именование

Парсер: YourParser (PascalCase)
Файл парсера: your_parser.py (snake_case)
Имя парсера: "your_parser" (snake_case)
Геттеры: get_data, get_full_data (snake_case)
Эндпоинты: /your_parser/upload, /your_parser/get_data

2. Структура данных

# Рекомендуемая структура возвращаемых данных
{
    "installation_id": {
        "data_type1": [
            {"field1": "value1", "field2": "value2"},
            {"field1": "value3", "field2": "value4"}
        ],
        "data_type2": [
            {"field1": "value5", "field2": "value6"}
        ]
    }
}

3. Обработка ошибок

try:
    # Ваш код
    result = some_operation()
    return result
except SpecificException as e:
    print(f"❌ Специфическая ошибка: {e}")
    return {}
except Exception as e:
    print(f"❌ Общая ошибка: {e}")
    raise

4. Логирование

# Используйте эмодзи для разных типов сообщений
print(f"🔍 DEBUG: Отладочная информация")
print(f"📁 INFO: Информационное сообщение")
print(f"✅ SUCCESS: Успешная операция")
print(f"⚠️ WARNING: Предупреждение")
print(f"❌ ERROR: Ошибка")

5. Валидация данных

# Всегда валидируйте входные параметры
validated_params = YourParserRequest(**params)

# Проверяйте наличие данных
if not data_source:
    print("⚠️ Нет данных в парсере")
    return {}

6. Документация

Добавляйте docstrings ко всем методам
Описывайте параметры в Pydantic схемах
Добавляйте примеры в FastAPI эндпоинты
Комментируйте сложную логику

📚 Примеры

Пример 1: Простой парсер

class SimpleParser(ParserPort):
    """Простой парсер для демонстрации"""
    
    name = "simple_parser"
    
    def __init__(self):
        super().__init__()
        self.register_getter("get_data", SimpleRequest, self._get_data_wrapper)
    
    def parse(self, file_path: str, params: dict) -> Dict[str, Any]:
        """Парсинг простого Excel файла"""
        df = pd.read_excel(file_path)
        return {"data": df.to_dict('records')}
    
    def _get_data_wrapper(self, params: dict) -> Dict[str, Any]:
        """Обертка для геттера"""
        data_source = self._get_data_source()
        return data_to_json(data_source)

Пример 2: Парсер с множественными геттерами

class MultiGetterParser(ParserPort):
    """Парсер с несколькими геттерами"""
    
    name = "multi_getter_parser"
    
    def __init__(self):
        super().__init__()
        # Регистрируем несколько геттеров
        self.register_getter("get_summary", SummaryRequest, self._get_summary_wrapper)
        self.register_getter("get_details", DetailsRequest, self._get_details_wrapper)
        self.register_getter("get_statistics", StatisticsRequest, self._get_statistics_wrapper)
    
    def _get_summary_wrapper(self, params: dict) -> Dict[str, Any]:
        """Геттер для получения сводки"""
        # Логика получения сводки
        pass
    
    def _get_details_wrapper(self, params: dict) -> Dict[str, Any]:
        """Геттер для получения деталей"""
        # Логика получения деталей
        pass
    
    def _get_statistics_wrapper(self, params: dict) -> Dict[str, Any]:
        """Геттер для получения статистики"""
        # Логика получения статистики
        pass

Пример 3: Парсер с фильтрацией

class FilteredParser(ParserPort):
    """Парсер с продвинутой фильтрацией"""
    
    def _filter_data(self, data_source: Dict[str, Any], params: FilterRequest) -> Dict[str, Any]:
        """Фильтрация данных по параметрам"""
        filtered_data = {}
        
        for key, value in data_source.items():
            # Фильтр по дате
            if params.start_date and value.get('date') < params.start_date:
                continue
            
            # Фильтр по типу
            if params.type and value.get('type') != params.type:
                continue
            
            # Фильтр по статусу
            if params.status and value.get('status') not in params.status:
                continue
            
            filtered_data[key] = value
        
        return filtered_data

🚀 Заключение

Это руководство покрывает все аспекты создания новых парсеров в системе NIN Excel Parsers API. Следуйте этим инструкциям для создания качественных, поддерживаемых и тестируемых парсеров.

Чек-лист для нового парсера:

Создана Pydantic схема
Создан класс парсера с геттерами
Парсер зарегистрирован в __init__.py
Парсер зарегистрирован в main.py
Добавлена логика в services.py
Созданы FastAPI эндпоинты
Добавлена вкладка в Streamlit
Создан тестовый скрипт
Проведено тестирование
Обновлена документация

Полезные ссылки:

Удачной разработки! 🚀

36 KiB Raw Permalink Blame History Unescape Escape