DFiP_Budget_planing/api/src/services/form_phase_service.py

from sqlalchemy.ext.asyncio import AsyncSession

from src.core.timezones import to_org_local_naive
from src.core.errors import AccessDeniedException, ValidationsError
from src.db.models.budget_form import BudgetForm
from src.db.models.role import UserRoleEnum
from src.repository.budget_form_repository import BudgetFormRepository
from src.db.models.app_user import AppUser
from src.db.models.form_phase import FormPhase
from src.domain.schemas import FormPhaseCreate, FormPhaseUpdate
from src.repository.form_phase_repository import FormPhaseRepository


# Код роли этапа (v3.role.code) по роли пользователя. Единственное место сопоставления,
# используется и FormPhaseService, и Form3PhaseService. Для ADMIN фильтр не нужен — его в
# словаре нет, role_code_for_user возвращает None.
ROLE_CODE_BY_USER_ROLE: dict[UserRoleEnum, str] = {
    UserRoleEnum.EXECUTOR_DFIP: "DFIP",
    UserRoleEnum.EXECUTOR_RF: "EXECUTOR_RF",
    UserRoleEnum.EXECUTOR_DURRS: "DURRS",
}


def role_code_for_user(user: AppUser) -> str | None:
    """Код роли этапа для фильтрации по пользователю; None — без фильтра (ADMIN)."""
    return ROLE_CODE_BY_USER_ROLE.get(user.role_id)


class FormPhaseService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.fp_repo = FormPhaseRepository(db)
        self.bf_repo = BudgetFormRepository(db)

    async def get_list(
        self,
        budget_form_id: int,
        user: AppUser,
        sheet: str | None = None,
        direction: str | None = None,
        phase_code: str | None = None,
    ) -> list[FormPhase]:
        return await self.fp_repo.get_list(
            budget_form_id=budget_form_id,
            sheet=sheet,
            direction=direction,
            phase_code=phase_code,
            role=role_code_for_user(user),
        )

    async def get(
        self, budget_form_id: int, sheet: str, direction: str | None, phase_code: str, user: AppUser
    ) -> FormPhase | None:
        return await self.fp_repo.get(
            budget_form_id=budget_form_id,
            sheet=sheet,
            direction=direction,
            phase_code=phase_code,
        )

    def _validate_form_phase(
            self,
            form_phase: FormPhaseCreate | FormPhaseUpdate,
            budget_form: BudgetForm,
            sheet: str | None = None,
            direction: str | None = None,
    ) -> None:
        errors = {}
        if budget_form.form_type.sheet_list_with_directions:
            if sheet:
                sheet = {"sheet_name": sheet, "direction": direction}
            else:
                sheet = {"sheet_name": form_phase.sheet, "direction": form_phase.direction}
            if sheet not in budget_form.form_type.sheet_list_with_directions:
                errors["sheet"] = [f"У формы нет листа {sheet['sheet_name']}/{sheet['direction']}"]
        if form_phase.column_keys and budget_form.form_type.section_list:
            for column_key in form_phase.column_keys:
                section = column_key.split(".")
                if len(section) == 2 and section[0] not in budget_form.form_type.section_list:
                    if "column_keys" in errors:
                        errors["column_keys"].append(f"Колонка {column_key} некорректна")
                    else:
                        errors["column_keys"] = [f"Колонка {column_key} некорректна"]
        if errors:
            raise ValidationsError(errors=errors)


    async def create(
            self,  
            body: FormPhaseCreate, 
            user: AppUser,
            budget_form_id: int | None = None,
            budget_form: BudgetForm | None = None,
            utc_offset: str | None = None,
    ) -> FormPhase:
        assert budget_form or budget_form_id
        if user.role_id != UserRoleEnum.ADMIN:
            raise AccessDeniedException
        
        if not budget_form or budget_form.form_type is None:
            budget_form = await self.bf_repo.get(
                budget_form_id=budget_form_id or budget_form.id,
                load_form_type=True,
            )
        self._validate_form_phase(
            form_phase=body,
            budget_form=budget_form,
        )
        opens_at = to_org_local_naive(body.opens_at, utc_offset)
        closes_at = to_org_local_naive(body.closes_at, utc_offset)
        return await self.fp_repo.create(
            budget_form_id=budget_form_id or budget_form.id,
            sheet=body.sheet,
            direction=body.direction,
            phase_code=body.phase_code,
            role=body.role,
            column_keys=body.column_keys,
            opens_at=opens_at,
            closes_at=closes_at,
        )

    async def update(
            self,
            sheet: str,
            direction: str | None,
            phase_code: str,
            body: FormPhaseUpdate,
            user: AppUser,
            budget_form_id: int | None = None,
            budget_form: BudgetForm | None = None,
            utc_offset: str | None = None,
    ) -> FormPhase:
        assert budget_form or budget_form_id
        if user.role_id != UserRoleEnum.ADMIN:
            raise AccessDeniedException
        
        if not budget_form or budget_form.form_type is None:
            budget_form = await self.bf_repo.get(
                budget_form_id=budget_form_id or budget_form.id,
                load_form_type=True,
            )
        self._validate_form_phase(
            form_phase=body,
            budget_form=budget_form,
            sheet=sheet,
            direction=direction,
        )

        data = body.model_dump(exclude_unset=True)
        for key in ("opens_at", "closes_at"):
            if data.get(key) is not None:
                data[key] = to_org_local_naive(data[key], utc_offset)
        if not data:
            return await self.fp_repo.get(
                budget_form_id=budget_form_id or budget_form.id,
                sheet=sheet,
                direction=direction,
                phase_code=phase_code,
            )
        return await self.fp_repo.update(
            budget_form_id=budget_form_id or budget_form.id,
            sheet=sheet,
            direction=direction,
            phase_code=phase_code,
            data=data,
        )

    async def delete(
        self, budget_form_id: int,
        sheet: str, 
        direction: str | None,
        phase_code: str, 
        user: AppUser,
    ) -> bool:
        return await self.fp_repo.delete(
            budget_form_id=budget_form_id,
            sheet=sheet,
            direction=direction,
            phase_code=phase_code,
        )