Merge pull request 'vsp-back: перенес функциональность справочников всп' (#40) from vsp-back into test

Reviewed-on: #40
Reviewed-by: Raykov-MS <RaykovMS@avt.rshb.ru>

api/src/api/v1/router.py

@@ -1,6 +1,8 @@
 from fastapi import APIRouter
 
-from src.api.v1 import auth, users, admin, audit, forms, form_phases, projects, export, org_unit
+from src.api.v1 import (
+    auth, users, admin, audit, forms, form_phases, projects, export, org_unit, vsp,
+)
 
 from src.api.v1 import websocket
 
@@ -16,3 +18,4 @@ api_router.include_router(form_phases.router)
 api_router.include_router(export.router)
 api_router.include_router(websocket.router)
 api_router.include_router(org_unit.router)
+api_router.include_router(vsp.router)

api/src/api/v1/vsp.py

@@ -0,0 +1,183 @@
+from datetime import date
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import StreamingResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.app_user import AppUser
+from src.api.v1.deps import require_admin, require_executor
+from src.db.session import get_db
+from src.domain.schemas import (
+    BaseListResponse,
+    BaseSingleResponse,
+    VSPCreate,
+    VSPExportRequest,
+    VSPInDB,
+    VSPUpdate,
+)
+from src.services.vsp_service import VSPService
+
+router = APIRouter(prefix="/dict/info", tags=["info"])
+
+
+@router.get("", response_model=BaseListResponse[VSPInDB])
+async def get_vsp(
+    registration_number: str | None = None,
+    address: str | None = None,
+    ssp_id: int | None = None,
+    open_date_start: date | None = None,
+    open_date_end: date | None = None,
+    location_form: str | None = None,
+    numbers_min: int | None = None,
+    numbers_max: int | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_executor),
+):
+    """Получение записей справочника INFO."""
+    service = VSPService(db)
+    result = await service.get_list(
+        user=current_user,
+        registration_number=registration_number,
+        address=address,
+        ssp_id=ssp_id,
+        open_date_start=open_date_start,
+        open_date_end=open_date_end,
+        location_form=location_form,
+        numbers_min=numbers_min,
+        numbers_max=numbers_max,
+        load_org_unit=True,
+        with_count=True,
+    )
+    return BaseListResponse(
+        success=True,
+        message="Список INFO",
+        result=[VSPInDB.model_validate(vsp) for vsp in result[1]],
+        count=result[0]
+    )
+
+
+@router.get("/dropdown", response_model=BaseListResponse[VSPInDB])
+async def get_vsp_dropdown(
+    ssp_id: int | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_executor),
+):
+    """Получение записей для выпадающего списка INFO."""
+    service = VSPService(db)
+    result = await service.get_dropdown(user=current_user, ssp_id=ssp_id, with_count=True)
+    return BaseListResponse(
+        success=True,
+        message="Список INFO для выпадающего списка",
+        result=[VSPInDB.model_validate(vsp) for vsp in result[1]],
+        count=result[0],
+    )
+
+
+@router.post("/export")
+async def export_info(
+    body: VSPExportRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_executor),
+):
+    """Экспорт справочника INFO."""
+    service = VSPService(db)
+    stream, filename = await service.export_xlsx(
+        user=current_user,
+        registration_number=body.registration_number,
+        address=body.address,
+        ssp_ids=body.ssp_ids,
+        open_date_start=body.open_date_start,
+        open_date_end=body.open_date_end,
+        location_form=body.location_form,
+        numbers_min=body.numbers_min,
+        numbers_max=body.numbers_max,
+    )
+    return StreamingResponse(
+        stream,
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
+
+
+@router.post("", response_model=BaseSingleResponse[VSPInDB], status_code=201)
+async def add_vsp(
+    body: VSPCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+):
+    """Добавление строки в справочник INFO."""
+    service = VSPService(db)
+    result = await service.create(
+        body,
+        current_user,
+        load_org_unit=True,
+    )
+    return BaseSingleResponse(
+        success=True,
+        message="Запись INFO создана",
+        result=VSPInDB.model_validate(result),
+    )
+
+
+@router.get("/{vsp_id}", response_model=BaseSingleResponse[VSPInDB])
+async def get_vsp_by_id(
+    vsp_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_executor),
+):
+    """Получение записи справочника INFO по ID."""
+    service = VSPService(db)
+    result = await service.get(vsp_id, user=current_user, load_org_unit=True)
+    if not result:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"ВСП не найден",
+        )
+    return BaseSingleResponse(
+        success=True,
+        message="Запись INFO",
+        result=VSPInDB.model_validate(result),
+    )
+
+
+@router.put("/{vsp_id}", response_model=BaseSingleResponse[VSPInDB])
+async def update_vsp(
+    vsp_id: int,
+    body: VSPUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_executor),
+):
+    """Редактирование строки справочника INFO."""
+    service = VSPService(db)
+    result = await service.update(
+        vsp_id,
+        body,
+        current_user,
+        load_org_unit=True,
+    )
+    if not result:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"ВСП не найден",
+        )
+    return BaseSingleResponse(
+        success=True,
+        message="Запись INFO обновлена",
+        result=VSPInDB.model_validate(result),
+    )
+
+
+@router.delete("/{vsp_id}", response_model=BaseSingleResponse[VSPInDB])
+async def delete_vsp(
+    vsp_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(require_admin),
+):
+    """Удаление строки справочника INFO."""
+    service = VSPService(db)
+    if await service.logical_delete(vsp_id, current_user):
+        return BaseSingleResponse(success=True, message="Запись INFO удалена")
+    raise HTTPException(
+        status_code=status.HTTP_404_NOT_FOUND,
+        detail=f"ВСП не найден",
+    ) 

api/src/db/models/__init__.py

@@ -22,7 +22,7 @@ from src.db.models.reserve import Reserve
 from src.db.models.rf_project_report import RfProjectReport
 from src.db.models.rf_project_report_line import RfProjectReportLine
 from src.db.models.rf_project_report_quarter import RfProjectReportQuarter
-from src.db.models.role import Role
+from src.db.models.role import Role, UserRoleEnum
 from src.db.models.security_detail import SecurityDetail
 from src.db.models.sequestration import Sequestration
 from src.db.models.user_org import UserOrg

api/src/db/models/org_unit.py

@@ -36,6 +36,6 @@ class OrgUnit(Base):
 #     projects: Mapped[list["Project"]] = relationship(  # type: ignore
 #         "Project", back_populates="org_unit"
 #     )
-#     vsps: Mapped[list["Vsp"]] = relationship("Vsp", back_populates="org_unit")  # type: ignore
+    vsps: Mapped[list["Vsp"]] = relationship("Vsp", back_populates="org_unit")  # type: ignore
     user_orgs: Mapped[list["UserOrg"]] = relationship("UserOrg", back_populates="org_unit")
     users: Mapped[list["AppUser"]] = relationship("AppUser", secondary="v3.user_org", back_populates="org_units")

api/src/db/models/vsp.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-from datetime import date
+from datetime import date, datetime
 
-from sqlalchemy import Date, ForeignKey, Integer, Numeric, String
+from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Integer, Numeric, String, Text, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from src.db.base import Base
@@ -22,8 +22,27 @@ class Vsp(Base):
     staff_count: Mapped[int | None] = mapped_column(Integer)
     total_area: Mapped[float | None] = mapped_column(Numeric)
     closed_at: Mapped[date | None] = mapped_column(Date)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    is_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
+    system_code: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_by: Mapped[int | None] = mapped_column(Integer, ForeignKey("v3.app_user.id"))
+    updated_by: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("v3.app_user.id"), nullable=True,
+    )
+    vsp_type: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+    location_form: Mapped[str | None] = mapped_column(String, nullable=True)
+    numbers: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    rent_contract_num: Mapped[str | None] = mapped_column(String, nullable=True)
+    rent_end_date : Mapped[date | None]= mapped_column(Date, nullable=True)
 
-#     org_unit = relationship("OrgUnit", back_populates="vsps")
+
+
+    org_unit = relationship("OrgUnit", back_populates="vsps")
 #     budget_lines: Mapped[list["BudgetLine"]] = relationship("BudgetLine", back_populates="vsp")  # type: ignore
 #     rent_details: Mapped[list["RentDetail"]] = relationship("RentDetail", back_populates="vsp")
 #     security_details: Mapped[list["SecurityDetail"]] = relationship("SecurityDetail", back_populates="vsp")

api/src/domain/enums.py

@@ -1,8 +1,23 @@
 import enum
 
 
+#ToDo: вероятно это теперь AuditEvent. А еще он нигде не юзается больше, вероятно, выпилить
 class AuditEventType(str, enum.Enum):
     LOGIN = "LOGIN"
     USER_CREATE = "USER_CREATE"
     USER_UPDATE = "USER_UPDATE"
     USER_DELETE = "USER_DELETE"
+    ROW_CREATE = "ROW_CREATE"
+    ROW_DELETE = "ROW_DELETE"
+    ACCESS_WINDOW_CHANGE = "ACCESS_WINDOW_CHANGE"
+    ACCESS_EXTEND = "ACCESS_EXTEND"
+    USER_ROLE_CHANGE = "USER_ROLE_CHANGE"
+    CREATE_TASK = "CREATE_TASK"
+    UPDATE_TASK = "UPDATE_TASK"
+    DELETE_TASK = "DELETE_TASK"
+    VSP_CREATE = "VSP_CREATE"
+    VSP_UPDATE = "VSP_UPDATE"
+    PROJECT_CREATE = "PROJECT_CREATE"
+    PROJECT_DELETE = "PROJECT_DELETE"
+    USER_ACCESS_GRANTED = "USER_ACCESS_GRANTED"
+    USER_ACCESS_REVOKED = "USER_ACCESS_REVOKED"

api/src/domain/schemas.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, date
 import enum
 from typing import Any, Dict, Generic, List, Literal, Optional, TypeVar
 from sqlalchemy.exc import MissingGreenlet
@@ -457,3 +457,159 @@ class AuditLogQueryParams(BaseModel):
     event_type: Optional[str] = Field(None, description="Тип события")
     date_from: Optional[datetime] = Field(None, description="Дата начала (ISO 8601)")
     date_to: Optional[datetime] = Field(None, description="Дата окончания (ISO 8601)")
+
+
+
+class VSPBase(BaseModel):
+    system_code: Optional[str] = Field(None, description="Системный код ВСП")
+    vsp_type: Optional[str] = Field(None, description="Вид ВСП")
+    registration_number: str = Field(
+        ..., 
+        description="Значение для выпадающего списка", 
+        validation_alias='reg_number',
+        serialization_alias='registration_number',
+    )
+    address: str = Field(..., description="Автоподставляемый адрес")
+    open_date: Optional[date] = Field(
+        None, 
+        description="Дата открытия", 
+        validation_alias='opened_at',
+        serialization_alias='open_date',
+    )
+    close_date: Optional[date] = Field(
+        None, 
+        description="Дата закрытия", 
+        validation_alias='closed_at',
+        serialization_alias='close_date',
+    )
+    approved_format: Optional[str] = Field(
+        None, 
+        description="Утвержденный формат", 
+        validation_alias='format',
+        serialization_alias='approved_format',
+    )
+    notes: Optional[str] = Field(None, description="Примечания")
+    location_form: Optional[str] = Field(None, description="Форма расположения")
+    numbers: Optional[int] = Field(None, description="Штатная численность")
+    area: Optional[float] = Field(
+        None, 
+        description="Арендная площадь", 
+        validation_alias='total_area',
+        serialization_alias='area',
+    )
+    rent_contract_num: Optional[str] = Field(None, description="Номер договора аренды")
+    rent_end_date: Optional[date] = Field(
+        None, description="Срок окончания договора аренды"
+    )
+    ssp_id: int = Field(
+        ..., 
+        description="ID ССП", 
+        validation_alias="branch_id",
+        serialization_alias="ssp_id",
+    )
+
+
+class VSPEditBase(BaseModel):
+    system_code: Optional[str] = Field(None, description="Системный код ВСП")
+    vsp_type: Optional[str] = Field(None, description="Вид ВСП")
+    reg_number: str = Field(
+        ..., 
+        description="Значение для выпадающего списка", 
+        validation_alias='registration_number',
+        serialization_alias='reg_number',
+    )
+    address: str = Field(..., description="Автоподставляемый адрес")
+    opened_at: Optional[date] = Field(
+        None, 
+        description="Дата открытия", 
+        validation_alias='open_date',
+        serialization_alias='opened_at',
+    )
+    closed_at: Optional[date] = Field(
+        None, 
+        description="Дата закрытия", 
+        validation_alias='close_date',
+        serialization_alias='closed_at',
+    )
+    format: Optional[str] = Field(
+        None, 
+        description="Утвержденный формат", 
+        validation_alias='approved_format',
+        serialization_alias='format',
+    )
+    notes: Optional[str] = Field(None, description="Примечания")
+    location_form: Optional[str] = Field(None, description="Форма расположения")
+    numbers: Optional[int] = Field(None, description="Штатная численность")
+    total_area: Optional[float] = Field(
+        None, 
+        description="Арендная площадь", 
+        validation_alias='area',
+        serialization_alias='total_area',
+    )
+    rent_contract_num: Optional[str] = Field(None, description="Номер договора аренды")
+    rent_end_date: Optional[date] = Field(
+        None, description="Срок окончания договора аренды"
+    )
+    branch_id: int = Field(
+        ...,
+        description="ID ССП", 
+        validation_alias="ssp_id",
+        serialization_alias="branch_id",
+    )
+
+
+class VSPCreate(VSPEditBase):
+    pass
+
+
+class VSPUpdate(VSPEditBase):
+    reg_number: str = Field(
+        None, 
+        description="Значение для выпадающего списка", 
+        validation_alias='registration_number',
+        serialization_alias='reg_number',
+    )
+    address: str = Field(None, description="Автоподставляемый адрес")
+    branch_id: int = Field(
+        None,
+        description="ID ССП", 
+        validation_alias="ssp_id",
+        serialization_alias="branch_id",
+    )
+
+
+class VSPInDB(VSPBase):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    is_active: bool
+    is_deleted: bool = False
+    created_by: int
+    updated_by: Optional[int] = None
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+    regional_branch: Optional[str] = Field(None, description="Региональный филиал")
+
+    @model_validator(mode='before')
+    @classmethod
+    def get_regional_branch(cls, data: Any) -> Any:
+
+        if isinstance(data, dict):
+            return data
+        result = vars(data)
+        result["regional_branch"] = None
+        try:
+            result["regional_branch"] = data.org_unit.title
+        except MissingGreenlet:
+            pass
+        return result
+
+
+class VSPExportRequest(BaseModel):
+    registration_number: Optional[str] = None
+    address: Optional[str] = None
+    ssp_ids: Optional[List[int]] = None
+    open_date_start: Optional[date] = None
+    open_date_end: Optional[date] = None
+    location_form: Optional[str] = None
+    numbers_min: Optional[int] = None
+    numbers_max: Optional[int] = None

api/src/repository/vsp_repository.py

@@ -0,0 +1,243 @@
+from datetime import date
+from typing import Optional
+
+from sqlalchemy import func, select, text, update
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import joinedload
+
+
+from src.db.models import Vsp
+from src.domain.schemas import VSPCreate
+
+
+class VSPRepository:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get(self, vsp_id: int, is_active: bool | None = None, load_org_unit: bool = False) -> Optional[Vsp]:
+        queries = [Vsp.id == vsp_id, Vsp.is_deleted.is_(False)]
+        if is_active is not None:
+            queries.append(Vsp.is_active == is_active)
+        if load_org_unit:
+            return (
+                (await self.db.execute(
+                    select(Vsp).where(*queries).options(joinedload(Vsp.org_unit)).limit(1))
+                ).scalars().first()
+            )
+        else:
+            return (
+                (await self.db.execute(select(Vsp).where(*queries).limit(1))).scalars().first()
+            )
+
+    async def get_list(
+        self,
+        registration_number: str | None = None,
+        address: str | None = None,
+        ssp_ids: list[int] | None = None,
+        open_date_start: date | None = None,
+        open_date_end: date | None = None,
+        location_form: str | None = None,
+        numbers_min: int | None = None,
+        numbers_max: int | None = None,
+        is_active: bool | None = True,
+        close_date_is_null: bool | None = None,
+        load_org_unit: bool = False,
+        with_count: bool = False,
+    ) -> list[Vsp] | tuple[int, list[Vsp]]:
+        
+        if with_count:
+            query = select(func.count().over().label("total_count"), Vsp)
+        else:
+            query = select(Vsp)
+        queries = [Vsp.is_deleted.is_(False)]
+        filters = [
+            ("registration_number", registration_number),
+            ("address", address),
+            ("location_form", location_form),
+            ("open_date_start", open_date_start),
+            ("open_date_end", open_date_end),
+            ("numbers_min", numbers_min),
+            ("numbers_max", numbers_max),
+            ("is_active", is_active),
+        ]
+        for filter_name, filter_value in filters:
+            # print(filter_name, filter_value)
+            match filter_name, filter_value:
+                case ("registration_number", str(value)) if value:
+                    queries.append(Vsp.reg_number.ilike(f"%{value}%"))
+                case ("address", str(value)) if value:
+                    queries.append(Vsp.address.ilike(f"%{value}%"))
+                case ("location_form", str(value)) if value:
+                    queries.append(Vsp.location_form.ilike(f"%{value}%"))
+                case ("open_date_start", value) if value is not None:
+                    queries.append(Vsp.opened_at >= value)
+                case ("open_date_end", value) if value is not None:
+                    queries.append(Vsp.opened_at <= value)
+                case ("numbers_min", value) if value is not None:
+                    queries.append(Vsp.numbers >= value)
+                case ("numbers_max", value) if value is not None:
+                    queries.append(Vsp.numbers <= value)
+                case ("is_active", value) if value is not None:
+                    queries.append(Vsp.is_active == value)
+                case _:
+                    continue
+
+        if close_date_is_null is True:
+            queries.append(Vsp.closed_at.is_(None))
+        elif close_date_is_null is False:
+            queries.append(Vsp.closed_at.is_not(None))
+
+        if ssp_ids is not None:
+            if not ssp_ids:
+                if with_count:
+                    return 0, []
+                return []
+            queries.append(Vsp.branch_id.in_(ssp_ids))
+
+        query = query.where(*queries)
+        if load_org_unit:
+            query = query.options(joinedload(Vsp.org_unit))
+        query = query.order_by(Vsp.id)
+        
+        if with_count:
+            result = (await self.db.execute(query)).all()
+            if result:
+                return result[0][0], [res[1] for res in result]
+            else:
+                if query.whereclause is not None:
+                    count_query = select(func.count(Vsp.id)).where(
+                        query.whereclause
+                    )
+                    return (await self.db.execute(count_query)).scalar(), []
+        else:
+            return (await self.db.execute(query)).scalars().all()
+        return (await self.db.execute(query.order_by(Vsp.id))).scalars().all()
+
+    async def create(self, payload: VSPCreate, created_by: int, load_org_unit: bool = False) -> Vsp:
+        data = payload.model_dump()
+        vsp_id = (
+            await self.db.execute(
+                text("""
+                    SELECT v3.add_vsp(
+                        :p_branch_id, :p_reg_number, :p_address, :p_format,
+                        :p_opened_at, :p_placement_type, :p_staff_count,
+                        :p_total_area, :p_closed_at, :p_is_active, :p_is_deleted,
+                        :p_system_code, :p_created_by, :p_vsp_type, :p_notes,
+                        :p_location_form, :p_numbers, :p_rent_contract_num,
+                        :p_rent_end_date
+                    )
+                """),
+                {
+                    "p_branch_id": data.get("branch_id"),
+                    "p_reg_number": data.get("reg_number"),
+                    "p_address": data.get("address"),
+                    "p_format": data.get("format"),
+                    "p_opened_at": data.get("opened_at"),
+                    "p_placement_type": data.get("placement_type"),
+                    "p_staff_count": data.get("staff_count"),
+                    "p_total_area": data.get("total_area"),
+                    "p_closed_at": data.get("closed_at"),
+                    "p_is_active": data.get("is_active", True),
+                    "p_is_deleted": data.get("is_deleted", False),
+                    "p_system_code": data.get("system_code"),
+                    "p_created_by": created_by,
+                    "p_vsp_type": data.get("vsp_type"),
+                    "p_notes": data.get("notes"),
+                    "p_location_form": data.get("location_form"),
+                    "p_numbers": data.get("numbers"),
+                    "p_rent_contract_num": data.get("rent_contract_num"),
+                    "p_rent_end_date": data.get("rent_end_date"),
+                }
+            )
+        ).scalar()
+
+        await self.db.flush()
+
+        if load_org_unit:
+            return (
+                await self.db.execute(
+                    select(Vsp).where(Vsp.id == vsp_id).options(joinedload(Vsp.org_unit)).limit(1)
+                )
+            ).scalars().first()
+        else:
+            return (
+                await self.db.execute(
+                    select(Vsp).where(Vsp.id == vsp_id).limit(1)
+                )
+            ).scalars().first()
+
+    async def update(self, vsp: Vsp, data: dict, updated_by: int, load_org_unit: bool = False) -> Vsp:
+        FIELD_TO_PARAM = {
+            "system_code": "p_system_code",
+            "vsp_type": "p_vsp_type",
+            "reg_number": "p_reg_number",
+            "address": "p_address",
+            "opened_at": "p_opened_at",
+            "closed_at": "p_closed_at",
+            "format": "p_format",
+            "notes": "p_notes",
+            "location_form": "p_location_form",
+            "numbers": "p_numbers",
+            "total_area": "p_total_area",
+            "rent_contract_num": "p_rent_contract_num",
+            "rent_end_date": "p_rent_end_date",
+            "branch_id": "p_branch_id",
+            "is_active": "p_is_active",
+            "is_deleted": "p_is_deleted",
+        }
+
+        params = {"p_vsp_id": vsp.id, "p_updated_by": updated_by}
+        for field, param in FIELD_TO_PARAM.items():
+            params[param] = data.get(field)
+        params.setdefault("p_placement_type", data.get("placement_type"))
+        params.setdefault("p_staff_count", data.get("staff_count"))
+        await self.db.execute(
+            text("""
+                SELECT v3.upd_vsp(
+                    :p_vsp_id, :p_branch_id, :p_reg_number, :p_address,
+                    :p_format, :p_opened_at, :p_placement_type, :p_staff_count,
+                    :p_total_area, :p_closed_at, :p_is_active, :p_is_deleted,
+                    :p_system_code, :p_updated_by, :p_vsp_type, :p_notes,
+                    :p_location_form, :p_numbers, :p_rent_contract_num,
+                    :p_rent_end_date
+                )
+            """),
+            params,
+        )
+
+        await self.db.flush()
+        await self.db.refresh(vsp)
+
+        if load_org_unit:
+            return (
+                await self.db.execute(
+                    select(Vsp).where(Vsp.id == vsp.id).options(joinedload(Vsp.org_unit)).limit(1)
+                )
+            ).scalars().first()
+        else:
+            return vsp
+
+    async def deactivate_by_close_date(self, as_of: date) -> int:
+        result = await self.db.execute(
+            update(Vsp)
+            .where(
+                Vsp.is_active.is_(True),
+                Vsp.closed_at.is_not(None),
+                Vsp.closed_at <= as_of,
+            )
+            .values(is_active=False, updated_at=func.now())
+        )
+        await self.db.commit()
+        return result.rowcount or 0
+
+    async def logical_delete(self, vsp_id: int, updated_by: int) -> bool:
+        result = (
+            await self.db.execute(
+                text("SELECT v3.del_vsp(:p_vsp_id, :p_deleted_by)"),
+                {"p_vsp_id": vsp_id, "p_deleted_by": updated_by},
+            )
+        ).scalar()
+        await self.db.flush()
+        return result is not None
+
+

api/src/services/vsp_service.py

@@ -0,0 +1,273 @@
+import asyncio
+import io
+from datetime import date, datetime
+from typing import Optional
+
+from openpyxl import Workbook
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.core.errors import AccessDeniedException, ValidationException
+from src.db.models import AppUser, UserRoleEnum, Vsp
+from src.domain.schemas import VSPCreate, VSPUpdate
+from src.repository.user_repository import UserRepository
+from src.repository.vsp_repository import VSPRepository
+
+_last_info_close_sync_run: Optional[date] = None
+_info_close_sync_lock = asyncio.Lock()
+
+
+class VSPService:
+    EXECUTOR_EDITABLE_FIELDS = {
+        "location_form",
+        "numbers",
+        "area",
+        "rent_contract_num",
+        "rent_end_date",
+    }
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+        self.vsp_repo = VSPRepository(db)
+        self.user_repo = UserRepository(db)
+
+    async def get(self, vsp_id: int, user: AppUser, load_org_unit: bool = False) -> Optional[Vsp]:
+        vsp = await self.vsp_repo.get(vsp_id=vsp_id, load_org_unit=load_org_unit)
+        if not vsp:
+            return None
+        if user.role_id == UserRoleEnum.ADMIN:
+            return vsp
+
+        ssp_ids = await self._get_scoped_ssp_ids(user=user)
+        if not ssp_ids or vsp.ssp_id not in set(ssp_ids):
+            raise AccessDeniedException()
+        return vsp
+
+    async def get_list(
+        self,
+        user: AppUser,
+        registration_number: str | None = None,
+        address: str | None = None,
+        ssp_id: int | None = None,
+        ssp_ids: list[int] | None = None,
+        open_date_start: date | None = None,
+        open_date_end: date | None = None,
+        location_form: str | None = None,
+        numbers_min: int | None = None,
+        numbers_max: int | None = None,
+        is_active: bool | None = None,
+        load_org_unit: bool = False,
+        with_count: bool = False,
+    ) -> list[Vsp] | tuple[int, list[Vsp]]:
+        await self._sync_closed_records()
+        scoped_ssp_ids = await self._get_scoped_ssp_ids(user=user)
+        requested_ssp_ids = list(dict.fromkeys(ssp_ids or []))
+        if ssp_id is not None and ssp_id not in requested_ssp_ids:
+            requested_ssp_ids.append(ssp_id)
+
+        effective_ssp_ids = scoped_ssp_ids
+        if requested_ssp_ids:
+            if scoped_ssp_ids is None:
+                effective_ssp_ids = requested_ssp_ids
+            else:
+                scoped_set = set(scoped_ssp_ids)
+                effective_ssp_ids = [item for item in requested_ssp_ids if item in scoped_set]
+                if not effective_ssp_ids:
+                    if with_count:
+                        return 0, []
+                    return []
+        return await self.vsp_repo.get_list(
+            registration_number=registration_number,
+            address=address,
+            ssp_ids=effective_ssp_ids,
+            open_date_start=open_date_start,
+            open_date_end=open_date_end,
+            location_form=location_form,
+            numbers_min=numbers_min,
+            numbers_max=numbers_max,
+            is_active=is_active,
+            load_org_unit=load_org_unit,
+            with_count=with_count,
+        )
+
+    async def get_dropdown(
+            self, 
+            user: AppUser, 
+            ssp_id: int | None = None,
+            with_count: bool = False,
+    ) -> list[Vsp]:
+        ssp_ids = await self._get_scoped_ssp_ids(user=user)
+
+        if ssp_id is not None:
+            if ssp_ids is None:
+                ssp_ids = [ssp_id]
+            elif ssp_id not in set(ssp_ids):
+                if with_count:
+                    return 0, []
+                return []
+            else:
+                ssp_ids = [ssp_id]
+
+        return await self.vsp_repo.get_list(
+            ssp_ids=ssp_ids,
+            is_active=True,
+            close_date_is_null=True,
+            with_count=with_count,
+        )
+
+    async def create(self, payload: VSPCreate, user: AppUser, load_org_unit: bool = False) -> Vsp:
+        if not self._can_modify(user):
+            raise AccessDeniedException()
+        try:
+            created = await self.vsp_repo.create(payload=payload, created_by=user.id, load_org_unit=load_org_unit)
+            if created.closed_at is not None and created.closed_at <= date.today():
+                created = await self.vsp_repo.update(
+                    vsp=created,
+                    data={"is_active": False},
+                    updated_by=user.id,
+                    load_org_unit=load_org_unit,
+                )
+            return created
+        except IntegrityError as exc:
+            raise ValidationException(description="Некорректные данные для записи ВСП") from exc
+    
+    async def export_xlsx(
+        self,
+        user: AppUser,
+        registration_number: str | None = None,
+        address: str | None = None,
+        ssp_ids: list[int] | None = None,
+        open_date_start: date | None = None,
+        open_date_end: date | None = None,
+        location_form: str | None = None,
+        numbers_min: int | None = None,
+        numbers_max: int | None = None,
+    ) -> tuple[io.BytesIO, str]:
+        data = await self.get_list(
+            user=user,
+            registration_number=registration_number,
+            address=address,
+            ssp_ids=ssp_ids,
+            open_date_start=open_date_start,
+            open_date_end=open_date_end,
+            location_form=location_form,
+            numbers_min=numbers_min,
+            numbers_max=numbers_max,
+            is_active=None,
+            load_org_unit=True,
+        )
+
+        workbook = Workbook()
+        sheet = workbook.active
+        sheet.title = "INFO"
+        sheet.append(
+            [
+                "id",
+                "ssp_id",
+                "system_code",
+                "regional_branch",
+                "vsp_type",
+                "registration_number",
+                "address",
+                "approved_format",
+                "open_date",
+                "close_date",
+                "notes",
+                "location_form",
+                "numbers",
+                "area",
+                "rent_contract_num",
+                "rent_end_date",
+                "is_active",
+                "is_deleted",
+                "created_at",
+                "updated_at",
+                "created_by",
+                "updated_by",
+            ]
+        )
+        for item in data:
+            sheet.append(
+                [
+                    item.id,
+                    item.branch_id,
+                    item.system_code or None,
+                    item.org_unit.title or None,
+                    item.vsp_type or None,
+                    item.reg_number,
+                    item.address,
+                    item.format or None,
+                    item.opened_at.isoformat() if item.opened_at else None,
+                    item.closed_at.isoformat() if item.closed_at else None,
+                    item.notes or None,
+                    item.location_form or None,
+                    item.numbers if item.numbers is not None else None,
+                    float(item.total_area) if item.total_area is not None else None,
+                    item.rent_contract_num or None,
+                    item.rent_end_date.isoformat() if item.rent_end_date else None,
+                    item.is_active,
+                    item.is_deleted,
+                    item.created_at.isoformat() if item.created_at else None,
+                    item.updated_at.isoformat() if item.updated_at else None,
+                    item.created_by,
+                    item.updated_by if item.updated_by is not None else None,
+                ]
+            )
+
+        stream = io.BytesIO()
+        workbook.save(stream)
+        workbook.close()
+        stream.seek(0)
+        file_name = f"info_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
+        return stream, file_name
+
+    async def update(self, vsp_id: int, payload: VSPUpdate, user: AppUser, load_org_unit: bool = False) -> Vsp | None:
+        vsp = await self.vsp_repo.get(vsp_id=vsp_id, is_active=None)
+        if not vsp:
+            return None
+
+        data = payload.model_dump(exclude_unset=True)
+        if not data:
+            return vsp
+
+        if user.role_id != UserRoleEnum.ADMIN:
+            ssp_ids = await self._get_scoped_ssp_ids(user=user)
+            if not ssp_ids or vsp.ssp_id not in set(ssp_ids):
+                raise AccessDeniedException()
+            if not set(data.keys()).issubset(self.EXECUTOR_EDITABLE_FIELDS):
+                raise AccessDeniedException()
+
+        effective_close_date = data.get("close_date", vsp.closed_at)
+        if effective_close_date is not None and effective_close_date <= date.today():
+            data["is_active"] = False
+
+        try:
+            return await self.vsp_repo.update(
+                vsp=vsp, data=data, updated_by=user.id, load_org_unit=load_org_unit,
+            )
+        except IntegrityError as exc:
+            raise ValidationException(description="Некорректные данные для записи ВСП") from exc
+
+    async def logical_delete(self, vsp_id: int, user: AppUser) -> bool:
+        if not self._can_modify(user):
+            raise AccessDeniedException()
+        return await self.vsp_repo.logical_delete(vsp_id=vsp_id, updated_by=user.id)
+
+    def _can_modify(self, user: AppUser) -> bool:
+        return user.role_id == UserRoleEnum.ADMIN
+
+    async def _get_scoped_ssp_ids(self, user: AppUser) -> list[int] | None:
+        if user.role_id == UserRoleEnum.ADMIN:
+            return None
+        return list(set(await self.user_repo.get_many_ssp_ids(user_id=user.id)))
+
+    async def _sync_closed_records(self) -> None:
+        global _last_info_close_sync_run
+        today = date.today()
+        if _last_info_close_sync_run is not None and _last_info_close_sync_run >= today:
+            return
+        async with _info_close_sync_lock:
+            if _last_info_close_sync_run is not None and _last_info_close_sync_run >= today:
+                return
+            await self.vsp_repo.deactivate_by_close_date(as_of=today)
+            _last_info_close_sync_run = today

api/tests/integration/test_vsp.py

@@ -0,0 +1,228 @@
+import uuid
+
+import pytest
+
+
+def _unique_reg_number() -> str:
+    return f"TEST_{uuid.uuid4().hex[:10]}"
+
+
+def test_vsp_list_smoke(client, admin_tokens, auth_headers):
+    response = client.get("/api/v1/dict/info", headers=auth_headers(admin_tokens))
+    assert response.status_code == 200
+    payload = response.json()
+    assert "result" in payload
+    assert "count" in payload
+    assert isinstance(payload["result"], list)
+
+
+def test_vsp_list_with_filters(client, admin_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/dict/info?ssp_id=1",
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert isinstance(payload["result"], list)
+
+
+def test_vsp_dropdown_smoke(client, admin_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/dict/info/dropdown", headers=auth_headers(admin_tokens)
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert "result" in payload
+    assert "count" in payload
+    assert isinstance(payload["result"], list)
+
+
+def test_vsp_dropdown_with_ssp_filter(client, admin_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/dict/info/dropdown?ssp_id=1",
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert isinstance(payload["result"], list)
+
+
+def test_vsp_get_by_id(client, admin_tokens, auth_headers):
+    response = client.get("/api/v1/dict/info/1", headers=auth_headers(admin_tokens))
+    assert response.status_code == 200
+    payload = response.json()
+    assert "result" in payload
+    assert isinstance(payload["result"], dict)
+    assert payload["result"]["id"] == 1
+
+
+def test_vsp_get_not_found(client, admin_tokens, auth_headers):
+    response = client.get(
+        "/api/v1/dict/info/99999", headers=auth_headers(admin_tokens)
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_vsp_create_smoke(client, admin_tokens, auth_headers):
+    reg_number = _unique_reg_number()
+    response = client.post(
+        "/api/v1/dict/info",
+        json={
+            "registration_number": reg_number,
+            "address": "Test Address",
+            "ssp_id": 1,
+        },
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 201
+    payload = response.json()
+    assert payload["success"] is True
+    assert payload["result"]["registration_number"] == reg_number
+    assert payload["result"]["address"] == "Test Address"
+    assert payload["result"]["ssp_id"] == 1
+
+
+def test_vsp_create_with_all_fields(client, admin_tokens, auth_headers):
+    reg_number = _unique_reg_number()
+    response = client.post(
+        "/api/v1/dict/info",
+        json={
+            "registration_number": reg_number,
+            "address": "Full Address",
+            "ssp_id": 1,
+            "system_code": "SYS001",
+            "vsp_type": "Тип 1",
+            "open_date": "2024-01-15",
+            "close_date": "2025-12-31",
+            "approved_format": "Стандарт",
+            "notes": "Тестовое примечание",
+            "location_form": "Встроенное",
+            "numbers": 10,
+            "area": 150.5,
+            "rent_contract_num": "Д-123/24",
+            "rent_end_date": "2025-12-31",
+        },
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 201
+    payload = response.json()
+    assert payload["success"] is True
+    assert payload["result"]["registration_number"] == reg_number
+    assert payload["result"]["system_code"] == "SYS001"
+    assert payload["result"]["vsp_type"] == "Тип 1"
+    assert payload["result"]["open_date"] == "2024-01-15"
+    assert payload["result"]["close_date"] == "2025-12-31"
+    assert payload["result"]["approved_format"] == "Стандарт"
+    assert payload["result"]["location_form"] == "Встроенное"
+    assert payload["result"]["numbers"] == 10
+    assert payload["result"]["area"] == 150.5
+    assert payload["result"]["rent_contract_num"] == "Д-123/24"
+    assert payload["result"]["rent_end_date"] == "2025-12-31"
+
+
+def test_vsp_create_forbidden_for_executor(client, isp_tokens, auth_headers):
+    response = client.post(
+        "/api/v1/dict/info",
+        json={
+            "registration_number": _unique_reg_number(),
+            "address": "Should Fail",
+            "ssp_id": 1,
+        },
+        headers=auth_headers(isp_tokens),
+    )
+    assert response.status_code == 403
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_vsp_update_smoke(client, admin_tokens, auth_headers):
+    reg_number = _unique_reg_number()
+    created = client.post(
+        "/api/v1/dict/info",
+        json={
+            "registration_number": reg_number,
+            "address": "Original Address",
+            "ssp_id": 1,
+        },
+        headers=auth_headers(admin_tokens),
+    )
+    assert created.status_code == 201
+    vsp_id = created.json()["result"]["id"]
+
+    new_address = f"Updated Address {uuid.uuid4().hex[:4]}"
+    response = client.put(
+        f"/api/v1/dict/info/{vsp_id}",
+        json={"address": new_address},
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["success"] is True
+    assert payload["result"]["address"] == new_address
+
+
+def test_vsp_update_not_found(client, admin_tokens, auth_headers):
+    response = client.put(
+        "/api/v1/dict/info/99999",
+        json={"address": "No Matter"},
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_vsp_delete_smoke(client, admin_tokens, auth_headers):
+    reg_number = _unique_reg_number()
+    created = client.post(
+        "/api/v1/dict/info",
+        json={
+            "registration_number": reg_number,
+            "address": "To Be Deleted",
+            "ssp_id": 1,
+        },
+        headers=auth_headers(admin_tokens),
+    )
+    assert created.status_code == 201
+    vsp_id = created.json()["result"]["id"]
+
+    response = client.delete(
+        f"/api/v1/dict/info/{vsp_id}",
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["success"] is True
+    assert payload["message"] == "Запись INFO удалена"
+
+
+def test_vsp_delete_not_found(client, admin_tokens, auth_headers):
+    response = client.delete(
+        "/api/v1/dict/info/99999",
+        headers=auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_vsp_delete_forbidden_for_executor(client, isp_tokens, auth_headers):
+    response = client.delete(
+        "/api/v1/dict/info/1",
+        headers=auth_headers(isp_tokens),
+    )
+    assert response.status_code == 403
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "message" in payload
+
+
+def test_vsp_requires_auth(client):
+    response = client.get("/api/v1/dict/info")
+    assert response.status_code == 403