service/src/services/template_loader.py - master - aastray/StudentPart

service/src/services/template_loader.py
213 строк8 KB
aastray
Реализован функционал загрузки файлов и просмотра документов, удалены базовые эндпоинты, обновлена документация
10 янв 2026, 15:38
10 янв 2026, 15:384d47c55
"""
Сервис для загрузки шаблонов документов в БД при инициализации.
Парсит шаблоны из папки docx_templates/ и сохраняет метаданные в БД.
"""
import os
from pathlib import Path
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from src.database.models.document_template import DocumentTemplate
from src.services.template_parser import parse_template


# Маппинг файлов шаблонов к типам
TEMPLATE_MAPPING = {
    "Договор на  практику.docx": {
        "template_type": "договор",
        "practice_type": None  # Универсальный шаблон
    },
    "Индивидуальное задание.docx": {
        "template_type": "индивидуальное_задание",
        "practice_type": None  # Универсальный шаблон
    },
    "Лист учета инструктажа.docx": {
        "template_type": "лист_учета_инструктажа",
        "practice_type": None  # Универсальный шаблон
    },
    "Направление на практику.docx": {
        "template_type": "направление",
        "practice_type": None  # Универсальный шаблон
    }
}


def get_templates_directory() -> Path:
    """
    Возвращает путь к папке с шаблонами.
    
    Returns:
        Path: Путь к папке docx_templates/
    """
    # Для Docker: /app/docx_templates
    docker_path = Path("/app/docx_templates")
    if docker_path.exists():
        return docker_path
    
    # Для локального запуска: ищем относительно текущего файла
    # service/src/services/template_loader.py -> service/src/services -> service/src -> service -> корень проекта
    current_file = Path(__file__)
    
    # Пытаемся найти корень проекта (где находится docx_templates)
    # Сначала проверяем на уровень выше service/
    service_dir = current_file.parent.parent.parent  # service/src/services -> service/src -> service
    project_root = service_dir.parent  # service -> корень проекта
    local_path = project_root / "docx_templates"
    
    if local_path.exists():
        return local_path
    
    # Если не найдено в корне, проверяем в текущей директории service
    service_local_path = service_dir / "docx_templates"
    if service_local_path.exists():
        return service_local_path
    
    # Возвращаем ожидаемый путь (для создания директории или ошибки)
    return local_path


async def load_template_to_db(
    db: AsyncSession,
    template_file_name: str,
    template_type: str,
    practice_type: Optional[str] = None
) -> DocumentTemplate:
    """
    Загружает шаблон в БД, парсит его и сохраняет метаданные.
    
    Args:
        db: Сессия БД
        template_file_name: Имя файла шаблона
        template_type: Тип шаблона ("договор", "индивидуальное_задание", и т.д.)
        practice_type: Вид практики (опционально)
        
    Returns:
        DocumentTemplate: Созданная или обновлённая запись в БД
    """
    templates_dir = get_templates_directory()
    template_path = templates_dir / template_file_name
    
    if not template_path.exists():
        raise FileNotFoundError(f"Шаблон не найден: {template_path}")
    
    # Парсим шаблон для получения полей
    parsed_data = parse_template(str(template_path))
    
    # Проверяем, существует ли уже такой шаблон
    stmt = select(DocumentTemplate).where(
        DocumentTemplate.template_type == template_type,
        DocumentTemplate.practice_type == practice_type
    )
    result = await db.execute(stmt)
    existing_template = result.scalar_one_or_none()
    
    if existing_template:
        # Обновляем существующий шаблон
        existing_template.template_file_path = str(template_path.relative_to(templates_dir.parent))
        existing_template.student_fields = parsed_data['student_fields']
        existing_template.system_fields = parsed_data['system_fields']
        await db.commit()
        await db.refresh(existing_template)
        return existing_template
    else:
        # Создаём новый шаблон
        new_template = DocumentTemplate(
            template_type=template_type,
            practice_type=practice_type,
            template_name=template_file_name.replace('.docx', ''),
            template_file_path=str(template_path.relative_to(templates_dir.parent)),
            student_fields=parsed_data['student_fields'],
            system_fields=parsed_data['system_fields']
        )
        db.add(new_template)
        await db.commit()
        await db.refresh(new_template)
        return new_template


async def load_all_templates(db: AsyncSession, force_reload: bool = False) -> list[DocumentTemplate]:
    """
    Загружает все шаблоны из папки docx_templates/ в БД.
    
    Args:
        db: Сессия БД
        force_reload: Если True, перезагружает все шаблоны даже если они уже есть в БД
        
    Returns:
        List[DocumentTemplate]: Список загруженных шаблонов
    """
    templates_dir = get_templates_directory()
    
    if not templates_dir.exists():
        raise FileNotFoundError(f"Папка с шаблонами не найдена: {templates_dir}")
    
    loaded_templates = []
    
    # Загружаем шаблоны согласно маппингу
    for template_file_name, config in TEMPLATE_MAPPING.items():
        template_path = templates_dir / template_file_name
        
        if template_path.exists():
            try:
                template = await load_template_to_db(
                    db=db,
                    template_file_name=template_file_name,
                    template_type=config["template_type"],
                    practice_type=config["practice_type"]
                )
                loaded_templates.append(template)
                print(f"✓ Загружен шаблон: {template_file_name}")
            except Exception as e:
                print(f"✗ Ошибка при загрузке шаблона {template_file_name}: {e}")
        else:
            print(f"⚠ Шаблон не найден: {template_file_name}")
    
    return loaded_templates


async def get_template_from_db(
    db: AsyncSession,
    template_type: str,
    practice_type: Optional[str] = None
) -> Optional[DocumentTemplate]:
    """
    Получает шаблон из БД по типу и виду практики.
    
    Args:
        db: Сессия БД
        template_type: Тип шаблона
        practice_type: Вид практики (опционально)
        
    Returns:
        Optional[DocumentTemplate]: Шаблон или None если не найден
    """
    stmt = select(DocumentTemplate).where(
        DocumentTemplate.template_type == template_type
    )
    
    if practice_type:
        stmt = stmt.where(DocumentTemplate.practice_type == practice_type)
    else:
        stmt = stmt.where(DocumentTemplate.practice_type.is_(None))
    
    result = await db.execute(stmt)
    return result.scalar_one_or_none()


if __name__ == "__main__":
    """
    Скрипт для загрузки шаблонов в БД.
    Запуск: python -m src.services.template_loader
    """
    import asyncio
    from src.database.db import AsyncSessionLocal
    
    async def main():
        async with AsyncSessionLocal() as session:
            try:
                templates = await load_all_templates(session, force_reload=True)
                print(f"\nВсего загружено шаблонов: {len(templates)}")
            except Exception as e:
                print(f"Ошибка при загрузке шаблонов: {e}")
    
    asyncio.run(main())