55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
"""B04 지표면 모델 확정 서비스."""
|
|
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
|
|
from B04_wf1_Surface.B04_wf1_Surface_Repository import (
|
|
confirm_surface_model,
|
|
list_surface_models,
|
|
)
|
|
from common_util.common_util_surface_confirmation import (
|
|
merge_surface_confirmation_params,
|
|
)
|
|
from common_util.common_util_workflow_state import complete_stage
|
|
|
|
|
|
async def find_surface_model_for_selection(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
selection: dict[str, Any],
|
|
) -> int:
|
|
"""필터·표현 기본값과 일치하는 COMPLETE 모델 ID를 찾는다."""
|
|
for model in await list_surface_models(connection, project_id):
|
|
generation_params = model.get("generation_params") or {}
|
|
if (
|
|
model["status"] == "COMPLETE"
|
|
and model["model_type"].lower() == str(selection["method"]).lower()
|
|
and str(generation_params.get("source_filter", "")).lower()
|
|
== str(selection["source_filter"]).lower()
|
|
):
|
|
return int(model["id"])
|
|
raise LookupError(
|
|
"자동 확정 기본값과 일치하는 지표면 모델이 없습니다: "
|
|
f"filter={selection['source_filter']}, method={selection['method']}"
|
|
)
|
|
|
|
|
|
async def confirm_surface_selection(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
model_id: int,
|
|
selection: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""모델 확정, 선택값 스냅샷, WF1 완료를 현재 트랜잭션에서 처리한다."""
|
|
await confirm_surface_model(connection, project_id, model_id)
|
|
normalized = await merge_surface_confirmation_params(
|
|
connection,
|
|
str(project_id),
|
|
selection,
|
|
)
|
|
async with connection.cursor() as cursor:
|
|
await complete_stage(cursor, str(project_id), 1)
|
|
return normalized
|