You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
171 lines
8.5 KiB
Python
171 lines
8.5 KiB
Python
from app.api.v1.module_energy.chapter.crud import ChapterCRUD
|
|
from app.api.v1.module_energy.variable.crud import VariableCRUD
|
|
from app.api.v1.module_energy.report_template.crud import EnergyReportTemplateCRUD
|
|
from app.api.v1.module_energy.variable.service import VariableService
|
|
from app.api.v1.module_rule.rule.crud import RuleCRUD
|
|
from app.core.base_schema import AuthSchema
|
|
from app.core.exceptions import CustomException
|
|
|
|
from .schema import TemplateChapterRuleSchema, TemplateConfigSaveSchema, TemplateSingleChapterConfigSaveSchema
|
|
|
|
|
|
class TemplateConfigService:
|
|
"""能源报告模板配置服务。"""
|
|
|
|
def __init__(self, auth: AuthSchema) -> None:
|
|
self.auth = auth
|
|
|
|
async def save(self, data: TemplateConfigSaveSchema) -> dict:
|
|
await self._validate_template(template_id=data.template_id)
|
|
await self._validate_rules(chapters=data.chapters)
|
|
await self._save_chapters(template_id=data.template_id, chapters=data.chapters)
|
|
variable_result = await VariableService(self.auth).save(data=data)
|
|
return {
|
|
"template_id": data.template_id,
|
|
"chapters": await self._list_chapters(template_id=data.template_id),
|
|
"variables": variable_result,
|
|
}
|
|
|
|
async def detail(self, template_id: int) -> dict:
|
|
await self._validate_template(template_id=template_id)
|
|
return {
|
|
"template_id": template_id,
|
|
"chapters": await self._list_chapters(template_id=template_id),
|
|
"variables": await VariableService(self.auth).list_by_template(template_id=template_id),
|
|
}
|
|
|
|
async def create_chapter_config(self, data: TemplateSingleChapterConfigSaveSchema) -> dict:
|
|
await self._validate_template(template_id=data.template_id)
|
|
await self._validate_rules(chapters=data.chapters)
|
|
chapter = await self._create_chapter(template_id=data.template_id, chapter=data.chapters[0])
|
|
await self._save_variables(
|
|
template_id=data.template_id,
|
|
chapter_name=chapter.chapter_name,
|
|
variables=data.variables,
|
|
)
|
|
return await VariableService(self.auth).list_by_chapter(chapter_id=chapter.id)
|
|
|
|
async def update_chapter_config(self, chapter_id: int, data: TemplateSingleChapterConfigSaveSchema) -> dict:
|
|
await self._validate_template(template_id=data.template_id)
|
|
await self._validate_rules(chapters=data.chapters)
|
|
chapter = await self._update_chapter(
|
|
template_id=data.template_id,
|
|
chapter_id=chapter_id,
|
|
chapter=data.chapters[0],
|
|
)
|
|
await self._save_variables(
|
|
template_id=data.template_id,
|
|
chapter_name=chapter.chapter_name,
|
|
variables=data.variables,
|
|
)
|
|
return await VariableService(self.auth).list_by_chapter(chapter_id=chapter.id)
|
|
|
|
async def _save_chapters(self, template_id: int, chapters: list[TemplateChapterRuleSchema]) -> None:
|
|
for chapter in chapters:
|
|
chapter_data = chapter.model_dump(exclude={"id"})
|
|
chapter_data["template_id"] = template_id
|
|
if chapter.id:
|
|
old_chapter = await ChapterCRUD(self.auth).get(id=chapter.id, template_id=template_id)
|
|
if not old_chapter:
|
|
raise CustomException(msg=f"章节规则标注不存在或不属于当前模板: {chapter.id}")
|
|
await ChapterCRUD(self.auth).update(id=chapter.id, data=chapter_data)
|
|
else:
|
|
await ChapterCRUD(self.auth).create(data=chapter_data)
|
|
|
|
async def _create_chapter(self, template_id: int, chapter: TemplateChapterRuleSchema):
|
|
if chapter.id:
|
|
raise CustomException(msg="新增章节标注时 chapters[0].id 不能传值")
|
|
chapter_data = chapter.model_dump(exclude={"id"})
|
|
chapter_data["template_id"] = template_id
|
|
return await ChapterCRUD(self.auth).create(data=chapter_data)
|
|
|
|
async def _update_chapter(self, template_id: int, chapter_id: int, chapter: TemplateChapterRuleSchema):
|
|
old_chapter = await ChapterCRUD(self.auth).get(id=chapter_id, template_id=template_id)
|
|
if not old_chapter:
|
|
raise CustomException(msg=f"章节规则标注不存在或不属于当前模板: {chapter_id}")
|
|
if chapter.id and chapter.id != chapter_id:
|
|
raise CustomException(msg="请求体 chapters[0].id 必须与路径标注ID一致")
|
|
chapter_data = chapter.model_dump(exclude={"id"})
|
|
return await ChapterCRUD(self.auth).update(id=chapter_id, data=chapter_data)
|
|
|
|
async def _save_variables(self, template_id: int, chapter_name: str, variables: list) -> None:
|
|
for variable in variables:
|
|
existing = await VariableCRUD(self.auth).get_list(
|
|
search={"template_id": template_id, "variable_name": variable.variable_name},
|
|
)
|
|
if existing:
|
|
await VariableCRUD(self.auth).hard_delete(ids=[record.id for record in existing])
|
|
for variable_chapter_name in self._resolve_variable_chapter_names(
|
|
current_chapter_name=chapter_name,
|
|
variable=variable,
|
|
):
|
|
await VariableCRUD(self.auth).create(
|
|
data={
|
|
"template_id": template_id,
|
|
"variable_name": variable.variable_name,
|
|
"chapter_name": variable_chapter_name,
|
|
}
|
|
)
|
|
|
|
def _collect_variable_chapter_names(self, chapter_name: str, variables: list) -> list[str]:
|
|
chapter_names: list[str] = []
|
|
for variable in variables:
|
|
chapter_names.extend(self._resolve_variable_chapter_names(current_chapter_name=chapter_name, variable=variable))
|
|
return self._unique_names(chapter_names)
|
|
|
|
def _resolve_variable_chapter_names(self, current_chapter_name: str, variable) -> list[str]:
|
|
chapter_names = list(variable.chapter_names)
|
|
if variable.checked is True and current_chapter_name not in chapter_names:
|
|
chapter_names.append(current_chapter_name)
|
|
if variable.checked is False:
|
|
chapter_names = [chapter_name for chapter_name in chapter_names if chapter_name != current_chapter_name]
|
|
return self._unique_names(chapter_names)
|
|
|
|
def _unique_names(self, names: list[str]) -> list[str]:
|
|
return list(dict.fromkeys(name.strip() for name in names if name.strip()))
|
|
|
|
async def _validate_variable_chapters(self, template_id: int, chapter_names: list[str]) -> None:
|
|
if not chapter_names:
|
|
return
|
|
chapters = await ChapterCRUD(self.auth).get_list(search={"template_id": template_id})
|
|
exists_names = {chapter.chapter_name for chapter in chapters}
|
|
invalid_names = [chapter_name for chapter_name in chapter_names if chapter_name not in exists_names]
|
|
if invalid_names:
|
|
raise CustomException(msg=f"变量关联章节不存在或不属于当前模板: {invalid_names}")
|
|
|
|
async def _validate_template(self, template_id: int) -> None:
|
|
template = await EnergyReportTemplateCRUD(self.auth).get(id=template_id)
|
|
if not template:
|
|
raise CustomException(msg="能源报告模板不存在")
|
|
|
|
async def _validate_rules(self, chapters: list[TemplateChapterRuleSchema]) -> None:
|
|
rule_ids = list(dict.fromkeys(chapter.rule_id for chapter in chapters))
|
|
if not rule_ids:
|
|
return
|
|
rules = await RuleCRUD(self.auth).get_list(search={"id": ("in", rule_ids)})
|
|
exists_ids = {rule.id for rule in rules}
|
|
invalid_ids = [rule_id for rule_id in rule_ids if rule_id not in exists_ids]
|
|
if invalid_ids:
|
|
raise CustomException(msg=f"关联规则不存在: {invalid_ids}")
|
|
|
|
async def _list_chapters(self, template_id: int) -> list[dict]:
|
|
chapters = await ChapterCRUD(self.auth).get_list(
|
|
search={"template_id": template_id},
|
|
order_by=[{"chapter_order": "asc"}, {"id": "asc"}],
|
|
)
|
|
result: list[dict] = []
|
|
for chapter in chapters:
|
|
rule_name = chapter.rule.name if hasattr(chapter, "rule") and chapter.rule else None
|
|
result.append(
|
|
{
|
|
"id": chapter.id,
|
|
"template_id": chapter.template_id,
|
|
"chapter_name": chapter.chapter_name,
|
|
"chapter_order": chapter.chapter_order,
|
|
"rule_id": chapter.rule_id,
|
|
"rule_name": rule_name,
|
|
"description": chapter.description,
|
|
}
|
|
)
|
|
return result
|