app/services/document.py - master - tsps/core

app/services/document.py
222 строки7 KB
Василий Петров
Документы в ЛК
21 мар 2026, 16:39
21 мар 2026, 16:39cc0cd1f
# Общий сервис для работы с документами как в LK, так и в админке
from __future__ import annotations

from typing import Optional, Tuple
from datetime import date as date_type

from fastapi import HTTPException, status, Response
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import nullslast

from app.models.document import Document
from app.models.case_entity import CaseEntity
from app.schemas.document import DocumentRead, DocumentReadFull


def _apply_document_filters(
    query,
    title: Optional[str] = None,
    date_from: Optional[date_type] = None,
    date_to: Optional[date_type] = None,
    type_id: Optional[int] = None,
    court_id: Optional[int] = None,
    case_number: Optional[str] = None,
):
    """
    Применяет общие фильтры к запросу документов.

    Приватная функция, используется внутри сервиса.
    """
    if title:
        query = query.filter(Document.title.ilike(f"%{title}%"))
    if date_from:
        query = query.filter(Document.date >= date_from)
    if date_to:
        query = query.filter(Document.date <= date_to)
    if type_id is not None:
        query = query.filter(Document.type_id == type_id)
    if court_id is not None:
        query = query.filter(Document.court_id == court_id)
    if case_number:
        query = query.join(Document.case_entity).filter(
            CaseEntity.number.ilike(f"%{case_number}%")
        )
    return query


def list_documents(
    db: Session,
    skip: int = 0,
    limit: int = 100,
    judge_id: Optional[int] = None,
    court_id: Optional[int] = None,
    title: Optional[str] = None,
    date_from: Optional[date_type] = None,
    date_to: Optional[date_type] = None,
    type_id: Optional[int] = None,
    case_number: Optional[str] = None,
) -> dict:
    """
    Получает список документов с фильтрацией и пагинацией.

    Аргументы:
        db: Сессия базы данных.
        skip: Пропуск записей (для пагинации).
        limit: Лимит записей (для пагинации).
        judge_id: Фильтр по судье (если None — без фильтра).
        court_id: Фильтр по суду (если None — без фильтра).
        title: Поиск по заголовку (частичное совпадение).
        date_from: Фильтр по дате от.
        date_to: Фильтр по дате до.
        type_id: Фильтр по типу документа.
        case_number: Поиск по номеру дела (частичное совпадение).

    Возвращает:
        dict в формате PaginationOut: {items, total, skip, limit}
    """
    # Базовый запрос с подгрузкой связанных объектов
    query = db.query(Document).options(
        joinedload(Document.type),
        joinedload(Document.case_entity),
        joinedload(Document.judge),
        joinedload(Document.court)
    )
    
    # Фильтр по принадлежности
    if judge_id is not None:
        query = query.filter(Document.judge_id == judge_id)

    # Общие фильтры
    query = _apply_document_filters(
        query,
        title=title,
        date_from=date_from,
        date_to=date_to,
        type_id=type_id,
        court_id=court_id,
        case_number=case_number,
    )

    # Получаем общее количество до пагинации
    total = query.count()

    # Сортировка: date ASC (NULLS LAST), затем id
    query = query.order_by(nullslast(Document.date.asc()), Document.id.asc())

    # Пагинация
    documents = query.offset(skip).limit(limit).all()

    return {
        "items": documents,
        "total": total,
        "skip": skip,
        "limit": limit,
    }


def get_document(
    db: Session,
    document_id: int,
    judge_id: Optional[int] = None,
) -> DocumentReadFull:
    """
    Получает документ по ID с проверкой прав доступа.

    Аргументы:
        db: Сессия базы данных.
        document_id: ID документа.
        judge_id: Требуемый ID судьи (если None — проверка не выполняется).

    Возвращает:
        DocumentReadFull

    Raises:
        HTTPException(404): Документ не найден.
        HTTPException(403): Доступ запрещён (документ принадлежит другому судье/суду).
    """
    document = (
        db.query(Document)
        .options(
            joinedload(Document.type),
            joinedload(Document.case_entity),
            joinedload(Document.judge),
            joinedload(Document.court),
            joinedload(Document.document_file)
        )
        .filter(Document.id == document_id)
        .first()
    )

    if not document:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Документ не найден"
        )

    # Проверка прав доступа
    if judge_id is not None and document.judge_id != judge_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Доступ запрещён: документ принадлежит другому судье"
        )

    return document


def get_document_file_data(
    db: Session,
    document_id: int,
    judge_id: Optional[int] = None,
) -> Tuple[bytes, str, str, str]:
    """
    Получает данные PDF-файла документа.

    Аргументы:
        db: Сессия базы данных.
        document_id: ID документа.
        judge_id: Требуемый ID судьи (если None — проверка не выполняется).

    Возвращает:
        Tuple[content, mime_type, filename_ascii, filename_utf8]

    Raises:
        HTTPException(404): Документ не найден.
        HTTPException(403): Доступ запрещён.
        HTTPException(204): Файл отсутствует (No Content).
    """
    document = (
        db.query(Document)
        .options(joinedload(Document.document_file))
        .filter(Document.id == document_id)
        .first()
    )

    if not document:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Документ не найден"
        )

    # Проверка прав доступа
    if judge_id is not None and document.judge_id != judge_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Доступ запрещён: документ принадлежит другому судье"
        )

    if not document.document_file:
        # Специальный случай: документ есть, но файла нет
        raise HTTPException(
            status_code=status.HTTP_204_NO_CONTENT,
            detail="Файл документа отсутствует"
        )

    filename_ascii = f"document_{document_id}.pdf"
    filename_utf8 = f"{document.title}.pdf" if document.title else filename_ascii

    return (
        document.document_file.file_data,
        document.document_file.mime_type,
        filename_ascii,
        filename_utf8,
    )