from typing import Any, Literal, NoReturn from pydantic import BaseModel from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from common.vo import CrudResponseModel, PageModel from exceptions.exception import ServiceException, ServiceWarning from module_admin.entity.vo.user_vo import CurrentUserModel from module_business.dao.master_dao import MasterDao from module_business.domain.test_batch_scope import TestBatchScope, assert_record_in_scope from module_business.entity.vo.master_vo import ( CustomerModel, CustomerPageQueryModel, DeleteCustomerModel, DeleteLineModel, DeleteSiteModel, DeleteStrainModel, DeleteTunnelModel, DeleteWarehouseModel, LineModel, LinePageQueryModel, SiteModel, SitePageQueryModel, StrainModel, StrainPageQueryModel, TunnelModel, TunnelPageQueryModel, WarehouseModel, WarehousePageQueryModel, ) from module_business.service.test_batch_access_service import TestBatchAccessService MasterResource = Literal['strain', 'customer', 'site', 'line', 'tunnel', 'warehouse'] _RESOURCE_CONFIG: dict[MasterResource, dict[str, Any]] = { 'strain': { 'label': '菌种', 'id_field': 'strain_id', 'code_field': 'strain_code', 'ids_field': 'strain_ids', 'vo': StrainModel, }, 'customer': { 'label': '客户', 'id_field': 'customer_id', 'code_field': 'customer_code', 'ids_field': 'customer_ids', 'vo': CustomerModel, }, 'site': { 'label': '客户现场', 'id_field': 'site_id', 'code_field': 'site_code', 'ids_field': 'site_ids', 'vo': SiteModel, }, 'line': { 'label': '生产线', 'id_field': 'line_id', 'code_field': 'line_code', 'ids_field': 'line_ids', 'vo': LineModel, }, 'tunnel': { 'label': '隧道', 'id_field': 'tunnel_id', 'code_field': 'tunnel_code', 'ids_field': 'tunnel_ids', 'vo': TunnelModel, }, 'warehouse': { 'label': '仓库', 'id_field': 'warehouse_id', 'code_field': 'warehouse_code', 'ids_field': 'warehouse_ids', 'vo': WarehouseModel, }, } class MasterService: """菌种、客户、现场、产线和仓库主数据服务层。""" @staticmethod async def _prepare_test_write( db: AsyncSession, current_user: CurrentUserModel, scope: TestBatchScope, ) -> None: if scope.mode != 'TEST': return TestBatchAccessService.assert_super_admin(current_user) await TestBatchAccessService.lock_active_batch(db, scope) @staticmethod async def _validate_parent( db: AsyncSession, resource: MasterResource, page_object: BaseModel, scope: TestBatchScope, ) -> None: if resource == 'site': customer = await MasterDao.get_customer(db, page_object.customer_id, scope) if customer is None: raise ServiceException(message='目标客户不存在') assert_record_in_scope(customer, scope, '目标客户') if customer.status != '0': raise ServiceException(message='目标客户已停用') elif resource == 'line': site = await MasterDao.get_site(db, page_object.site_id, scope) if site is None: raise ServiceException(message='目标客户现场不存在') assert_record_in_scope(site, scope, '目标客户现场') if site.status != '0': raise ServiceException(message='目标客户现场已停用') elif resource == 'tunnel': site = await MasterDao.get_site(db, page_object.site_id, scope) if site is None: raise ServiceException(message='目标客户现场不存在') assert_record_in_scope(site, scope, '目标客户现场') if site.status != '0': raise ServiceException(message='目标客户现场已停用') @staticmethod async def _find_duplicate( db: AsyncSession, resource: MasterResource, page_object: BaseModel, ) -> Any | None: if resource == 'site': return await MasterDao.find_site_by_code( db, page_object.site_code, page_object.customer_id, ) if resource == 'line': return await MasterDao.find_line_by_code( db, page_object.line_code, page_object.site_id, ) if resource == 'tunnel': return await MasterDao.find_tunnel_by_code( db, page_object.tunnel_code, page_object.site_id, ) code_field = _RESOURCE_CONFIG[resource]['code_field'] return await getattr(MasterDao, f'find_{resource}_by_code')( db, getattr(page_object, code_field), ) @classmethod async def _validate_unique( cls, db: AsyncSession, resource: MasterResource, page_object: BaseModel, current_id: int | None = None, ) -> None: duplicate = await cls._find_duplicate(db, resource, page_object) if duplicate is None: return config = _RESOURCE_CONFIG[resource] duplicate_id = getattr(duplicate, config['id_field'], None) if current_id is None or duplicate_id != current_id: code = getattr(page_object, config['code_field']) raise ServiceException(message=f'{config["label"]}编码{code}已存在') @classmethod async def _list( cls, db: AsyncSession, resource: MasterResource, query_object: BaseModel, scope: TestBatchScope, is_page: bool, ) -> PageModel | list[BaseModel]: result = await getattr(MasterDao, f'list_{resource}')(db, query_object, scope, is_page) config = _RESOURCE_CONFIG[resource] records = result.rows if isinstance(result, PageModel) else result rows = [config['vo'].model_validate(record) for record in records] if isinstance(result, PageModel): return result.model_copy(update={'rows': rows}) return rows @classmethod async def _detail( cls, db: AsyncSession, resource: MasterResource, resource_id: int, scope: TestBatchScope, ) -> BaseModel: config = _RESOURCE_CONFIG[resource] record = await getattr(MasterDao, f'get_{resource}')(db, resource_id, scope) if record is None: raise ServiceException(message=f'{config["label"]}不存在') assert_record_in_scope(record, scope, config['label']) return config['vo'].model_validate(record) @classmethod async def _add( cls, db: AsyncSession, resource: MasterResource, page_object: BaseModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: config = _RESOURCE_CONFIG[resource] try: await cls._prepare_test_write(db, current_user, scope) await cls._validate_parent(db, resource, page_object, scope) await cls._validate_unique(db, resource, page_object) safe_object = page_object.model_copy(update={config['id_field']: None}) await getattr(MasterDao, f'add_{resource}')(db, safe_object, scope.batch_id) await db.commit() return CrudResponseModel(is_success=True, message='新增成功') except IntegrityError: await db.rollback() raise ServiceException(message=f'新增{config["label"]}失败,编码已存在或关联数据无效') from None except Exception: await db.rollback() raise @classmethod async def _edit( cls, db: AsyncSession, resource: MasterResource, page_object: BaseModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: config = _RESOURCE_CONFIG[resource] try: await cls._prepare_test_write(db, current_user, scope) resource_id = getattr(page_object, config['id_field']) if resource_id is None: raise ServiceException(message=f'{config["label"]}ID不能为空') current = await getattr(MasterDao, f'get_{resource}')( db, resource_id, scope, for_update=scope.mode == 'TEST', ) if current is None: raise ServiceException(message=f'{config["label"]}不存在') assert_record_in_scope(current, scope, config['label']) await cls._validate_parent(db, resource, page_object, scope) await cls._validate_unique(db, resource, page_object, resource_id) await getattr(MasterDao, f'update_{resource}')(db, page_object, scope) await db.commit() return CrudResponseModel(is_success=True, message='更新成功') except IntegrityError: await db.rollback() raise ServiceException(message=f'修改{config["label"]}失败,编码已存在或关联数据无效') from None except Exception: await db.rollback() raise @staticmethod async def reject_physical_delete() -> NoReturn: raise ServiceWarning('', '不支持单条物理删除,请通过测试数据批次清理') @classmethod async def list_strain( cls, db: AsyncSession, query_object: StrainPageQueryModel, scope: TestBatchScope, is_page: bool = True, ) -> PageModel | list[StrainModel]: return await cls._list(db, 'strain', query_object, scope, is_page) @classmethod async def detail_strain(cls, db: AsyncSession, strain_id: int, scope: TestBatchScope) -> StrainModel: return await cls._detail(db, 'strain', strain_id, scope) @classmethod async def add_strain( cls, db: AsyncSession, strain: StrainModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._add(db, 'strain', strain, current_user, scope) @classmethod async def edit_strain( cls, db: AsyncSession, strain: StrainModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._edit(db, 'strain', strain, current_user, scope) @classmethod async def remove_strain( cls, db: AsyncSession, value: int | DeleteStrainModel, ) -> NoReturn: return await cls.reject_physical_delete() @classmethod async def list_customer( cls, db: AsyncSession, query_object: CustomerPageQueryModel, scope: TestBatchScope, is_page: bool = True, ) -> PageModel | list[CustomerModel]: return await cls._list(db, 'customer', query_object, scope, is_page) @classmethod async def detail_customer(cls, db: AsyncSession, customer_id: int, scope: TestBatchScope) -> CustomerModel: return await cls._detail(db, 'customer', customer_id, scope) @classmethod async def add_customer( cls, db: AsyncSession, customer: CustomerModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._add(db, 'customer', customer, current_user, scope) @classmethod async def edit_customer( cls, db: AsyncSession, customer: CustomerModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._edit(db, 'customer', customer, current_user, scope) @classmethod async def remove_customer( cls, db: AsyncSession, value: int | DeleteCustomerModel, ) -> NoReturn: return await cls.reject_physical_delete() @classmethod async def list_site( cls, db: AsyncSession, query_object: SitePageQueryModel, scope: TestBatchScope, is_page: bool = True, ) -> PageModel | list[SiteModel]: return await cls._list(db, 'site', query_object, scope, is_page) @classmethod async def detail_site(cls, db: AsyncSession, site_id: int, scope: TestBatchScope) -> SiteModel: return await cls._detail(db, 'site', site_id, scope) @classmethod async def add_site( cls, db: AsyncSession, site: SiteModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._add(db, 'site', site, current_user, scope) @classmethod async def edit_site( cls, db: AsyncSession, site: SiteModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._edit(db, 'site', site, current_user, scope) @classmethod async def remove_site( cls, db: AsyncSession, value: int | DeleteSiteModel, ) -> NoReturn: return await cls.reject_physical_delete() @classmethod async def list_line( cls, db: AsyncSession, query_object: LinePageQueryModel, scope: TestBatchScope, is_page: bool = True, ) -> PageModel | list[LineModel]: return await cls._list(db, 'line', query_object, scope, is_page) @classmethod async def detail_line(cls, db: AsyncSession, line_id: int, scope: TestBatchScope) -> LineModel: return await cls._detail(db, 'line', line_id, scope) @classmethod async def add_line( cls, db: AsyncSession, line: LineModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._add(db, 'line', line, current_user, scope) @classmethod async def edit_line( cls, db: AsyncSession, line: LineModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._edit(db, 'line', line, current_user, scope) @classmethod async def remove_line( cls, db: AsyncSession, value: int | DeleteLineModel, ) -> NoReturn: return await cls.reject_physical_delete() @classmethod async def list_tunnel( cls, db: AsyncSession, query_object: TunnelPageQueryModel, scope: TestBatchScope, is_page: bool = True, ) -> PageModel | list[TunnelModel]: return await cls._list(db, 'tunnel', query_object, scope, is_page) @classmethod async def detail_tunnel(cls, db: AsyncSession, tunnel_id: int, scope: TestBatchScope) -> TunnelModel: return await cls._detail(db, 'tunnel', tunnel_id, scope) @classmethod async def add_tunnel( cls, db: AsyncSession, tunnel: TunnelModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._add(db, 'tunnel', tunnel, current_user, scope) @classmethod async def edit_tunnel( cls, db: AsyncSession, tunnel: TunnelModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._edit(db, 'tunnel', tunnel, current_user, scope) @classmethod async def remove_tunnel( cls, db: AsyncSession, value: int | DeleteTunnelModel, ) -> NoReturn: return await cls.reject_physical_delete() @classmethod async def list_warehouse( cls, db: AsyncSession, query_object: WarehousePageQueryModel, scope: TestBatchScope, is_page: bool = True, ) -> PageModel | list[WarehouseModel]: return await cls._list(db, 'warehouse', query_object, scope, is_page) @classmethod async def detail_warehouse(cls, db: AsyncSession, warehouse_id: int, scope: TestBatchScope) -> WarehouseModel: return await cls._detail(db, 'warehouse', warehouse_id, scope) @classmethod async def add_warehouse( cls, db: AsyncSession, warehouse: WarehouseModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._add(db, 'warehouse', warehouse, current_user, scope) @classmethod async def edit_warehouse( cls, db: AsyncSession, warehouse: WarehouseModel, current_user: CurrentUserModel, scope: TestBatchScope, ) -> CrudResponseModel: return await cls._edit(db, 'warehouse', warehouse, current_user, scope) @classmethod async def remove_warehouse( cls, db: AsyncSession, value: int | DeleteWarehouseModel, ) -> NoReturn: return await cls.reject_physical_delete()