DFiP_Budget_planing/api/src/services/summary_service.py

import copy

from sqlalchemy.ext.asyncio import AsyncSession

from src.db.models.app_user import AppUser
from src.db.models.role import UserRoleEnum
from src.repository.summary_repository import SummaryRepository
from src.repository.user_repository import UserRepository


EMPTY_SVOD2_FACETS = {
    "total": 0,
    "search_total": 0,
    "smeta_type": [],
    "smeta_direction": [],
    "org_name": [],
}


class SummaryService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.summary_repo = SummaryRepository(db)
        self.user_repo = UserRepository(db)

    async def get_svod1_rows(
        self,
        year: int,
        sheet: str,
        user: AppUser,
        org_ids: list[int] | None = None,
    ) -> list[tuple]:
        org_ids = await self._allowed_org_ids(user, org_ids)
        if org_ids == []:
            return []
        return await self.summary_repo.get_svod1_rows(
            year=year,
            sheet=sheet,
            org_ids=org_ids,
        )

    async def get_svod2_rows(
        self,
        year: int,
        sheet: str,
        user: AppUser,
        org_ids: list[int] | None = None,
        offset: int | None = None,
        limit: int | None = None,
        sort_by: str = "org_name",
        sort_direction: str = "asc",
    ) -> list[tuple]:
        org_ids = await self._allowed_org_ids(user, org_ids)
        if org_ids == []:
            return []
        return await self.summary_repo.get_svod2_rows(
            year=year,
            sheet=sheet,
            org_ids=org_ids,
            offset=offset,
            limit=limit,
            sort_by=sort_by,
            sort_direction=sort_direction,
        )

    async def get_svod2_page(
        self,
        year: int,
        sheet: str,
        user: AppUser,
        org_ids: list[int] | None = None,
        offset: int | None = None,
        limit: int | None = None,
        sort_by: str = "org_name",
        sort_direction: str = "asc",
        search: str | None = None,
        smeta_type: str | None = None,
        smeta_direction: str | None = None,
        org_name: str | None = None,
        view: str = "full",
    ) -> tuple[list[tuple], int]:
        org_ids = await self._allowed_org_ids(user, org_ids)
        if org_ids == []:
            return [], 0

        async def fetch(page_offset: int | None, page_limit: int | None):
            return await self.summary_repo.get_svod2_page(
                year=year,
                sheet=sheet,
                org_ids=org_ids,
                offset=page_offset,
                limit=page_limit,
                sort_by=sort_by,
                sort_direction=sort_direction,
                search=search,
                smeta_type=smeta_type,
                smeta_direction=smeta_direction,
                org_name=org_name,
                compact=view == "compact",
            )

        rows, total = await fetch(offset, limit)
        if not rows and offset:
            # Страница за концом выборки: total берётся из первой строки, которой нет.
            _, total = await fetch(0, 1)
        return rows, total

    async def get_svod2_facets(
        self,
        year: int,
        sheet: str,
        user: AppUser,
        org_ids: list[int] | None = None,
        search: str | None = None,
        smeta_type: str | None = None,
        smeta_direction: str | None = None,
        org_name: str | None = None,
    ) -> dict:
        org_ids = await self._allowed_org_ids(user, org_ids)
        if org_ids == []:
            return copy.deepcopy(EMPTY_SVOD2_FACETS)
        return await self.summary_repo.get_svod2_facets(
            year=year,
            sheet=sheet,
            org_ids=org_ids,
            search=search,
            smeta_type=smeta_type,
            smeta_direction=smeta_direction,
            org_name=org_name,
        )

    async def get_svod2_go_export_rows(
        self,
        year: int,
        user: AppUser,
        org_ids: list[int] | None = None,
        sort_by: str = "org_name",
        sort_direction: str = "asc",
    ) -> list[tuple]:
        org_ids = await self._allowed_org_ids(user, org_ids)
        if org_ids == []:
            return []
        return await self.summary_repo.get_svod2_go_export_rows(
            year=year,
            org_ids=org_ids,
            sort_by=sort_by,
            sort_direction=sort_direction,
        )

    async def _allowed_org_ids(
        self,
        user: AppUser,
        requested_org_ids: list[int] | None,
    ) -> list[int] | None:
        if user.role_id == UserRoleEnum.ADMIN:
            return requested_org_ids

        db_user = await self.user_repo.get(user_id=user.id, load_orgs=True)
        if not db_user:
            return []

        allowed_org_ids = {org.id for org in db_user.org_units}
        if requested_org_ids is None:
            return list(allowed_org_ids)
        return [org_id for org_id in requested_org_ids if org_id in allowed_org_ids]