csg-server/app/api/v1/module_energy/report_generation/service.py

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}"