from app.api.v1.module_common.file.crud import FileCRUD from app.api.v1.module_common.file.service import FileService from app.api.v1.module_energy.report_template.crud import EnergyReportTemplateCRUD from app.api.v1.module_system.user.crud import UserCRUD from app.core.base_schema import AuthSchema, PageResultSchema from app.core.exceptions import CustomException from .crud import EnergyReportGenerationCRUD, EnergyReportGenerationLogCRUD from .schema import ( EnergyReportGenerationCreateSchema, EnergyReportGenerationLogCreateSchema, EnergyReportGenerationLogOutSchema, EnergyReportGenerationLogQueryParam, EnergyReportGenerationOutSchema, EnergyReportGenerationQueryParam, EnergyReportGenerationStatsSchema, EnergyReportGenerationUpdateSchema, ) ENERGY_REPORT_GENERATION_BUSINESS_TYPE = "energy_report_generation" GENERATION_LOG_ACTION_NAMES = { "create_task": "创建生成任务", "start_generation": "开始生成", "generation_success": "已生成", "generation_failed": "生成失败", } class EnergyReportGenerationService: """能源报告生成任务服务。""" def __init__(self, auth: AuthSchema) -> None: self.auth = auth async def detail(self, id: int) -> EnergyReportGenerationOutSchema: generation = await EnergyReportGenerationCRUD(self.auth).get_or_404(id=id) return self._to_out_schema(generation) async def page( self, page_no: int, page_size: int, search: EnergyReportGenerationQueryParam | None = None, order_by: list[dict[str, str]] | None = None, ) -> dict: search_dict = vars(search) if search else None stats = await self._stats(search_dict=self._build_stats_search(search_dict)) offset = (page_no - 1) * page_size page_result = await EnergyReportGenerationCRUD(self.auth).page( offset=offset, limit=page_size, order_by=order_by or [{"id": "desc"}], search=search_dict, out_schema=EnergyReportGenerationOutSchema, ) items = [self._fill_out_dict(item) for item in page_result.items] return { "page_no": page_result.page_no, "page_size": page_result.page_size, "total": page_result.total, "has_next": page_result.has_next, "stats": stats.model_dump(), "items": items, } async def create(self, data: EnergyReportGenerationCreateSchema) -> EnergyReportGenerationOutSchema: await self._validate_template(template_id=data.template_id, report_type=data.report_type) await self._validate_file(file_id=data.file_id) attachment_file = await self._validate_attachment_file( attachment_file_id=data.attachment_file_id, file_id=data.file_id, ) await self._validate_generator(generator_id=data.generator_id) create_data = data.model_dump() create_data["attachment_url"] = attachment_file.file_url if attachment_file else None generation = await EnergyReportGenerationCRUD(self.auth).create(data=create_data) if data.file_id: await self._bind_file(file_id=data.file_id, generation_id=generation.id) if data.attachment_file_id: await self._bind_file(file_id=data.attachment_file_id, generation_id=generation.id) await EnergyReportGenerationLogService(self.auth).create_system_log( report_generation_id=generation.id, generation_name=generation.generation_name, action_type="create_task", description=f"创建生成任务：{generation.generation_name}", ) return self._to_out_schema(generation) async def update(self, id: int, data: EnergyReportGenerationUpdateSchema) -> EnergyReportGenerationOutSchema: old_generation = await EnergyReportGenerationCRUD(self.auth).get_or_404(id=id, msg="更新失败，能源报告生成任务不存在") await self._validate_template(template_id=data.template_id, report_type=data.report_type) await self._validate_file(file_id=data.file_id, current_generation_id=id) if data.file_id and old_generation.attachment_file_id and old_generation.attachment_file_id == data.file_id: raise CustomException(msg="更新失败，生成源文档不能使用原附件文件") if old_generation.file_id and old_generation.file_id == data.attachment_file_id: raise CustomException(msg="更新失败，附件不能使用原生成源文档文件") attachment_file = await self._validate_attachment_file( attachment_file_id=data.attachment_file_id, file_id=data.file_id, current_generation_id=id, ) await self._validate_generator(generator_id=data.generator_id) update_data = data.model_dump(exclude_unset=True) update_data["attachment_url"] = attachment_file.file_url if attachment_file else None generation = await EnergyReportGenerationCRUD(self.auth).update(id=id, data=update_data) if old_generation.file_id and old_generation.file_id != data.file_id: await self._delete_file_record(file_id=old_generation.file_id) if data.file_id: await self._bind_file(file_id=data.file_id, generation_id=id) if data.attachment_file_id: await self._bind_file(file_id=data.attachment_file_id, generation_id=id) if old_generation.attachment_file_id and old_generation.attachment_file_id != data.attachment_file_id: await self._delete_file_record(file_id=old_generation.attachment_file_id) return self._to_out_schema(generation) async def delete(self, ids: list[int]) -> None: if len(ids) < 1: raise CustomException(msg="删除失败，删除对象不能为空") generations = await EnergyReportGenerationCRUD(self.auth).get_list(search={"id": ("in", ids)}) generation_map = {generation.id: generation for generation in generations} for generation_id in ids: if generation_id not in generation_map: raise CustomException(msg="删除失败，能源报告生成任务不存在") for generation in generations: if generation.file_id: await self._unbind_file(file_id=generation.file_id, generation_id=generation.id) await EnergyReportGenerationCRUD(self.auth).delete(ids=ids) async def _stats(self, search_dict: dict | None = None) -> EnergyReportGenerationStatsSchema: records = await EnergyReportGenerationCRUD(self.auth).get_list(search=search_dict or {}) return EnergyReportGenerationStatsSchema( submitted_count=len(records), generating_count=len([record for record in records if record.status == 1]), generated_count=len([record for record in records if record.status == 2]), pending_count=len([record for record in records if record.status == 0]), failed_count=len([record for record in records if record.status == 3]), ) def _build_stats_search(self, search_dict: dict | None = None) -> dict: if not search_dict: return {} stats_keys = {"created_time", "updated_time", "tenant_id", "created_id", "updated_id"} return {key: value for key, value in search_dict.items() if key in stats_keys and value is not None} def _to_out_schema(self, generation) -> EnergyReportGenerationOutSchema: generation_out = EnergyReportGenerationOutSchema.model_validate(generation) if generation_out.template: generation_out.template_name = generation_out.template.template_name if generation_out.file: generation_out.file_url = generation_out.file.file_url generation_out.original_name = generation_out.file.original_name generation_out.report_file_url = generation_out.file_url generation_out.report_file_original_name = generation_out.original_name if generation_out.attachment_file: generation_out.attachment_url = generation_out.attachment_file.file_url generation_out.attachment_original_name = generation_out.attachment_file.original_name generation_out.attachment_file_url = generation_out.attachment_url generation_out.attachment_file_original_name = generation_out.attachment_original_name if hasattr(generation, "generator") and generation.generator: generation_out.generator_name = generation.generator.name return generation_out def _fill_out_dict(self, item: dict) -> dict: template_info = item.get("template") or {} file_info = item.get("file") or {} attachment_file_info = item.get("attachment_file") or {} generator_info = item.get("generator") or {} item["file"] = file_info or None item["attachment_file"] = attachment_file_info or None item["template_name"] = template_info.get("template_name") item["file_url"] = file_info.get("file_url") item["original_name"] = file_info.get("original_name") item["report_file_url"] = item["file_url"] item["report_file_original_name"] = item["original_name"] item["attachment_url"] = attachment_file_info.get("file_url") or item.get("attachment_url") item["attachment_original_name"] = attachment_file_info.get("original_name") item["attachment_file_url"] = item["attachment_url"] item["attachment_file_original_name"] = item["attachment_original_name"] item["generator_name"] = generator_info.get("name") return item async def _validate_template(self, template_id: int, report_type: int) -> None: template = await EnergyReportTemplateCRUD(self.auth).get(id=template_id) if not template: raise CustomException(msg="能源报告模板不存在") if template.report_type != report_type: raise CustomException(msg="报告类型与模板类型不一致") async def _validate_file(self, file_id: int | None, current_generation_id: int | None = None) -> None: if file_id is None: return 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_GENERATION_BUSINESS_TYPE: raise CustomException(msg="文件已绑定到其他业务类型") if file_record.business_id and file_record.business_id != current_generation_id: raise CustomException(msg="文件已绑定到其他能源报告生成任务") async def _validate_attachment_file( self, attachment_file_id: int | None, file_id: int | None, current_generation_id: int | None = None, ): if attachment_file_id is None: return None if file_id and attachment_file_id == file_id: raise CustomException(msg="附件文件不能和生成源文档相同") file_record = await FileCRUD(self.auth).get(id=attachment_file_id) if not file_record: raise CustomException(msg="附件文件记录不存在") if file_record.business_type and file_record.business_type != ENERGY_REPORT_GENERATION_BUSINESS_TYPE: raise CustomException(msg="附件文件已绑定到其他业务类型") if file_record.business_id and file_record.business_id != current_generation_id: raise CustomException(msg="附件文件已绑定到其他能源报告生成任务") return file_record async def _validate_generator(self, generator_id: int | None) -> None: if generator_id is None: return generator = await UserCRUD(self.auth).get(id=generator_id) if not generator: raise CustomException(msg="生成人不存在") async def _bind_file(self, file_id: int, generation_id: int) -> None: await FileCRUD(self.auth).set( ids=[file_id], business_type=ENERGY_REPORT_GENERATION_BUSINESS_TYPE, business_id=generation_id, ) async def _unbind_file(self, file_id: int, generation_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_GENERATION_BUSINESS_TYPE and file_record.business_id == generation_id: await FileCRUD(self.auth).set(ids=[file_id], business_id=None) async def _delete_file_record(self, file_id: int) -> None: await FileService.delete_service(auth=self.auth, ids=[file_id], delete_object=True) class EnergyReportGenerationLogService: """能源报告生成操作记录服务。""" def __init__(self, auth: AuthSchema) -> None: self.auth = auth async def detail(self, id: int) -> EnergyReportGenerationLogOutSchema: log = await EnergyReportGenerationLogCRUD(self.auth).get_or_404(id=id, msg="操作记录不存在") return self._to_out_schema(log) async def page( self, page_no: int, page_size: int, search: EnergyReportGenerationLogQueryParam | None = None, order_by: list[dict[str, str]] | None = None, ) -> PageResultSchema[EnergyReportGenerationLogOutSchema]: search_dict = vars(search) if search else {} page_result = await EnergyReportGenerationLogCRUD(self.auth).page( offset=(page_no - 1) * page_size, limit=page_size, order_by=order_by or [{"id": "desc"}], search=search_dict, out_schema=EnergyReportGenerationLogOutSchema, ) page_result.items = [self._fill_out_dict(item) for item in page_result.items] return page_result async def create(self, data: EnergyReportGenerationLogCreateSchema) -> EnergyReportGenerationLogOutSchema: generation = await EnergyReportGenerationCRUD(self.auth).get_or_404( id=data.report_generation_id, msg="新增操作记录失败，能源报告生成任务不存在", ) log = await self.create_system_log( report_generation_id=generation.id, generation_name=generation.generation_name, action_type=data.action_type, description=data.description or self._default_description(generation.generation_name, data.action_type), ) return self._to_out_schema(log) async def delete(self, ids: list[int]) -> None: if len(ids) < 1: raise CustomException(msg="删除失败，删除对象不能为空") await EnergyReportGenerationLogCRUD(self.auth).delete(ids=ids) async def create_system_log(self, report_generation_id: int, generation_name: str, action_type: str, description: str | None = None): action_name = GENERATION_LOG_ACTION_NAMES.get(action_type) if not action_name: raise CustomException(msg="操作类型不支持") return await EnergyReportGenerationLogCRUD(self.auth).create( data=EnergyReportGenerationLogCreateSchema( report_generation_id=report_generation_id, action_type=action_type, # type: ignore[arg-type] description=description, generation_name=generation_name, action_name=action_name, ), ) def _to_out_schema(self, log) -> EnergyReportGenerationLogOutSchema: log_out = EnergyReportGenerationLogOutSchema.model_validate(log) log_out.operator_id = log_out.created_id if getattr(log, "created_by", None): log_out.operator_name = log.created_by.name log_out.action_name = GENERATION_LOG_ACTION_NAMES.get(log.action_type, log.action_type) return log_out def _fill_out_dict(self, item: dict) -> dict: created_by = item.get("created_by") or {} item["operator_id"] = item.get("created_id") item["operator_name"] = created_by.get("name") item["action_name"] = GENERATION_LOG_ACTION_NAMES.get(item.get("action_type"), item.get("action_name")) return item def _default_description(self, generation_name: str, action_type: str) -> str: action_name = GENERATION_LOG_ACTION_NAMES.get(action_type, action_type) return f"{action_name}：{generation_name}"