src/json_mapper.py - master - serohvostov/radio_proxy

src/json_mapper.py
233 строки11 KB
serohvostov
jinja support
23 янв 2026, 20:29
23 янв 2026, 20:29f85cc6a
"""
JSON Mapper для режима Grid
Создает JSON-файлы с настраиваемым маппингом полей из пакета 1211
"""

import json
import os
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
from dataclasses import asdict
from pathlib import Path

from jinja2 import Environment, FileSystemLoader, Template, TemplateNotFound, TemplateSyntaxError, UndefinedError, StrictUndefined

from .models import MetadataPacket, JSONMappingError


class Jinja2Renderer:
    """Рендерер для шаблонов Jinja2"""
    
    def __init__(self, template_source: Optional[str] = None, template_file: Optional[str] = None, custom_filters: Optional[Dict[str, Callable]] = None):
        """
        Инициализация рендерера с поддержкой inline шаблонов и файлов.
        
        Args:
            template_source: Строка с inline шаблоном Jinja2 (опционально)
            template_file: Путь к файлу с шаблоном (опционально)
            custom_filters: Словарь пользовательских фильтров для регистрации (опционально)
            
        Raises:
            JSONMappingError: Если не указан ни template_source, ни template_file
        """
        if template_source is None and template_file is None:
            raise JSONMappingError(
                'Необходимо указать либо template_source (inline шаблон), либо template_file (путь к файлу)'
            )
        
        self.template_source = template_source
        self.template_file = template_file
        self.template: Optional[Template] = None
        
        # Создание Jinja2 Environment с оптимизациями
        if template_file:
            # Для файловых шаблонов используем FileSystemLoader
            template_path = Path(template_file)
            template_dir = template_path.parent
            template_name = template_path.name
            
            self.env = Environment(
                loader=FileSystemLoader(str(template_dir)),
                autoescape=False,  # Отключаем автоэкранирование для JSON
                cache_size=400,  # Размер кеша шаблонов
                auto_reload=True,  # Автоматическая перезагрузка при изменении файлов
                undefined=StrictUndefined  # Строгая проверка неопределенных переменных
            )
        else:
            # Для inline шаблонов создаем Environment без FileSystemLoader
            self.env = Environment(
                autoescape=False,  # Отключаем автоэкранирование для JSON
                cache_size=400,  # Размер кеша шаблонов
                undefined=StrictUndefined  # Строгая проверка неопределенных переменных
            )
        
        # Регистрируем пользовательские фильтры ДО компиляции шаблона
        if custom_filters:
            for filter_name, filter_func in custom_filters.items():
                self.env.filters[filter_name] = filter_func
        
        # Теперь загружаем и компилируем шаблон
        if template_file:
            # Загружаем шаблон из файла
            try:
                self.template = self.env.get_template(template_name)
            except TemplateNotFound as e:
                raise JSONMappingError(f'Шаблон не найден: {template_file}') from e
            except TemplateSyntaxError as e:
                raise JSONMappingError(
                    f'Синтаксическая ошибка в шаблоне {template_file} на строке {e.lineno}: {e.message}'
                ) from e
        else:
            # Загружаем inline шаблон
            try:
                self.template = self.env.from_string(template_source)
            except TemplateSyntaxError as e:
                raise JSONMappingError(
                    f'Синтаксическая ошибка в inline шаблоне на строке {e.lineno}: {e.message}'
                ) from e
    
    def render(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """
        Рендерить шаблон с заданным контекстом.
        
        Args:
            context: Словарь с данными для шаблона
            
        Returns:
            Dict[str, Any]: Результат рендеринга (Python объект)
            
        Raises:
            JSONMappingError: При ошибках рендеринга или валидации
        """
        try:
            # Рендеринг шаблона с контекстом
            rendered_str = self.template.render(context)
            
            # Парсинг результата в Python объект (dict/list)
            try:
                result = json.loads(rendered_str)
            except json.JSONDecodeError as e:
                raise JSONMappingError(
                    f'Результат рендеринга не является валидным JSON: {e.msg} на позиции {e.pos}'
                ) from e
            
            # Валидация результата как JSON (должен быть dict или list)
            if not isinstance(result, (dict, list)):
                raise JSONMappingError(
                    f'Результат рендеринга должен быть объектом (dict) или массивом (list), '
                    f'получен: {type(result).__name__}'
                )
            
            return result
            
        except UndefinedError as e:
            raise JSONMappingError(f'Неопределенная переменная в шаблоне: {e.message}') from e
        except TemplateSyntaxError as e:
            raise JSONMappingError(
                f'Синтаксическая ошибка в шаблоне на строке {e.lineno}: {e.message}'
            ) from e
        except JSONMappingError:
            # Пробрасываем наши собственные ошибки
            raise
        except Exception as e:
            raise JSONMappingError(f'Ошибка рендеринга шаблона: {str(e)}') from e
    
    def register_filter(self, name: str, filter_func: Callable) -> None:
        """
        Зарегистрировать пользовательский фильтр.
        
        Args:
            name: Имя фильтра
            filter_func: Функция фильтра
        """
        self.env.filters[name] = filter_func
    
    def load_template(self, template_path: str) -> None:
        """
        Загрузить шаблон из файла.
        
        Args:
            template_path: Путь к файлу шаблона
            
        Raises:
            JSONMappingError: Если файл не найден или содержит ошибки
        """
        try:
            # Обновляем путь к файлу шаблона
            self.template_file = template_path
            
            # Создаем новый Environment с FileSystemLoader
            template_path_obj = Path(template_path)
            template_dir = template_path_obj.parent
            template_name = template_path_obj.name
            
            self.env = Environment(
                loader=FileSystemLoader(str(template_dir)),
                autoescape=False,
                cache_size=400,
                auto_reload=True,
                undefined=StrictUndefined  # Строгая проверка неопределенных переменных
            )
            
            # Загружаем шаблон
            self.template = self.env.get_template(template_name)
            
        except TemplateNotFound as e:
            raise JSONMappingError(f'Шаблон не найден: {template_path}') from e
        except TemplateSyntaxError as e:
            raise JSONMappingError(
                f'Синтаксическая ошибка в шаблоне {template_path} на строке {e.lineno}: {e.message}'
            ) from e
        except Exception as e:
            raise JSONMappingError(f'Ошибка загрузки шаблона {template_path}: {str(e)}') from e


class JSONMapper:
    """Класс для создания JSON-файлов с использованием Jinja2 шаблонов"""
    
    def __init__(self, template_path: str):
        """
        Инициализация JSON маппера
        
        Args:
            template_path: Путь к Jinja2 шаблону
        """
        self.template_path = template_path
        
        # Получаем все пользовательские фильтры
        from .jinja2_filters import CUSTOM_FILTERS
        
        # Создание Jinja2Renderer с фильтрами
        self.renderer = Jinja2Renderer(
            template_file=template_path,
            custom_filters=CUSTOM_FILTERS
        )
    
    def create_json(self, metadata: Dict[str, Any]) -> Dict[str, Any]:
        """
        Создать JSON с использованием Jinja2 шаблона
        
        Args:
            metadata: Словарь с данными из пакета 1211
            
        Returns:
            Dict[str, Any]: JSON объект с маппированными полями
        """
        return self.renderer.render(metadata)
    
    def create_json_from_metadata_packet(self, metadata_packet: MetadataPacket, **extra_context) -> Dict[str, Any]:
        """
        Создать JSON из объекта MetadataPacket
        
        Args:
            metadata_packet: Объект пакета метаданных
            **extra_context: Дополнительные переменные для контекста шаблона (например, unique_id)
            
        Returns:
            Dict[str, Any]: JSON объект
        """
        # Преобразуем MetadataPacket в словарь
        metadata_dict = asdict(metadata_packet)
        # Добавляем дополнительные переменные в контекст
        metadata_dict.update(extra_context)
        return self.create_json(metadata_dict)