284 lines
11 KiB
Python
284 lines
11 KiB
Python
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",
|
|
"total_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.branch_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:
|
|
message = str(getattr(exc, "orig", exc) or exc).strip()
|
|
if "vsp_reg_number_unique" in message:
|
|
raise ValidationException(description="Указанный номер уже существует")
|
|
if "chk_closed_after_opened" in message:
|
|
raise ValidationException(description="Дата закрытия не может быть меньше даты открытия")
|
|
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.branch_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:
|
|
message = str(getattr(exc, "orig", exc) or exc).strip()
|
|
if "vsp_reg_number_unique" in message:
|
|
raise ValidationException(description="Указанный номер уже существует")
|
|
if "chk_closed_after_opened" in message:
|
|
raise ValidationException(description="Дата закрытия не может быть меньше даты открытия")
|
|
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
|