from sqlalchemy import func, select from app.api.v1.module_common.file.crud import FileCRUD from app.api.v1.module_energy.chapter.model import EnergyReportChapterModel from app.core.base_schema import AuthSchema from app.core.exceptions import CustomException from .crud import EnergyReportTemplateCRUD from .model import EnergyReportTemplateModel from .schema import ( EnergyReportTemplateCreateSchema, EnergyReportTemplateOutSchema, EnergyReportTemplateQueryParam, EnergyReportTemplateUpdateSchema, ) ENERGY_REPORT_TEMPLATE_BUSINESS_TYPE = "energy_report_template" class EnergyReportTemplateService: """能源报告模板服务。""" def __init__(self, auth: AuthSchema) -> None: self.auth = auth async def detail(self, id: int) -> EnergyReportTemplateOutSchema: template = await EnergyReportTemplateCRUD(self.auth).get_or_404(id=id) return self._to_out_schema(template) async def page( self, page_no: int, page_size: int, search: EnergyReportTemplateQueryParam | None = None, order_by: list[dict[str, str]] | None = None, ) -> dict: offset = (page_no - 1) * page_size page_result = await EnergyReportTemplateCRUD(self.auth).page( offset=offset, limit=page_size, order_by=order_by or [{"order": "asc"}, {"id": "desc"}], search=vars(search) if search else None, out_schema=EnergyReportTemplateOutSchema, ) annotation_count_map = await self._annotation_count_map( template_ids=[item.get("id") for item in page_result.items if item.get("id")], ) for item in page_result.items: file_info = item.get("file") or {} item["file_url"] = file_info.get("file_url") item["original_name"] = file_info.get("original_name") item["object_name"] = file_info.get("object_name") item["annotation_count"] = annotation_count_map.get(item.get("id"), 0) return page_result async def create(self, data: EnergyReportTemplateCreateSchema) -> EnergyReportTemplateOutSchema: await self._validate_unique_name(template_name=data.template_name, tenant_id=self._current_tenant_id()) await self._validate_file(file_id=data.file_id) template = await EnergyReportTemplateCRUD(self.auth).create(data=data) await self._bind_file(file_id=data.file_id, template_id=template.id) return self._to_out_schema(template) async def update(self, id: int, data: EnergyReportTemplateUpdateSchema) -> EnergyReportTemplateOutSchema: old_template = await EnergyReportTemplateCRUD(self.auth).get_or_404(id=id, msg="更新失败，能源报告模板不存在") await self._validate_unique_name(template_name=data.template_name, exclude_id=id, tenant_id=old_template.tenant_id) await self._validate_file(file_id=data.file_id, current_template_id=id) template = await EnergyReportTemplateCRUD(self.auth).update(id=id, data=data) if old_template.file_id != data.file_id: await self._unbind_file(file_id=old_template.file_id, template_id=id) await self._bind_file(file_id=data.file_id, template_id=id) return self._to_out_schema(template) async def delete(self, ids: list[int]) -> None: if len(ids) < 1: raise CustomException(msg="删除失败，删除对象不能为空") templates = await EnergyReportTemplateCRUD(self.auth).get_list(search={"id": ("in", ids)}) template_map = {template.id: template for template in templates} for template_id in ids: if template_id not in template_map: raise CustomException(msg="删除失败，能源报告模板不存在") for template in templates: await self._unbind_file(file_id=template.file_id, template_id=template.id) await EnergyReportTemplateCRUD(self.auth).delete(ids=ids) def _to_out_schema(self, template) -> EnergyReportTemplateOutSchema: template_out = EnergyReportTemplateOutSchema.model_validate(template) if template_out.file: template_out.file_url = template_out.file.file_url template_out.original_name = template_out.file.original_name template_out.object_name = template_out.file.object_name return template_out async def _validate_unique_name( self, template_name: str, exclude_id: int | None = None, tenant_id: int | None = None, ) -> None: if self.auth.db is None: raise CustomException(msg="数据库会话不存在") sql = select(EnergyReportTemplateModel.id).where( EnergyReportTemplateModel.template_name == template_name, EnergyReportTemplateModel.is_deleted == False, ) if tenant_id is not None: sql = sql.where(EnergyReportTemplateModel.tenant_id == tenant_id) if exclude_id is not None: sql = sql.where(EnergyReportTemplateModel.id != exclude_id) result = await self.auth.db.execute(sql.limit(1)) if result.scalar_one_or_none() is not None: raise CustomException(msg="能源报告模板名称已存在") async def _annotation_count_map(self, template_ids: list[int]) -> dict[int, int]: if not template_ids: return {} if self.auth.db is None: raise CustomException(msg="数据库会话不存在") result = await self.auth.db.execute( select(EnergyReportChapterModel.template_id, func.count(EnergyReportChapterModel.id)) .where( EnergyReportChapterModel.template_id.in_(template_ids), EnergyReportChapterModel.is_deleted == False, ) .group_by(EnergyReportChapterModel.template_id) ) return {template_id: count for template_id, count in result.all()} def _current_tenant_id(self) -> int | None: if not self.auth.user: return self.auth.tenant_id return self.auth.tenant_id or self.auth.user.tenant_id async def _validate_file(self, file_id: int, current_template_id: int | None = None) -> None: file_record = await FileCRUD(self.auth).get(id=file_id) if not file_record: raise CustomException(msg="文件记录不存在") if file_record.business_type and file_record.business_type != ENERGY_REPORT_TEMPLATE_BUSINESS_TYPE: raise CustomException(msg="文件已绑定到其他业务类型") if file_record.business_id and file_record.business_id != current_template_id: raise CustomException(msg="文件已绑定到其他能源报告模板") async def _bind_file(self, file_id: int, template_id: int) -> None: await FileCRUD(self.auth).set( ids=[file_id], business_type=ENERGY_REPORT_TEMPLATE_BUSINESS_TYPE, business_id=template_id, ) async def _unbind_file(self, file_id: int, template_id: int) -> None: file_record = await FileCRUD(self.auth).get(id=file_id) if not file_record: return if file_record.business_type == ENERGY_REPORT_TEMPLATE_BUSINESS_TYPE and file_record.business_id == template_id: await FileCRUD(self.auth).set(ids=[file_id], business_id=None)