eomsangdonandClaude Opus 5
5666086c6e
- 완료 카드에서도 [파일 선택]을 보임. 올리는 중에만 감춤. 교체 확인 모달과 교체 업로드 규칙은 이미 있던 것을 그대로 씀. - 업로드 관문이 전처리 1단계 IN_PROGRESS 를 무조건 「분석 중」으로 읽어, 자동 확정이 보류된 프로젝트는 영영 새 자료를 받지 못했음. 전처리 진행 파일이 awaiting_confirmation·completed·failed 로 끝나 있으면 받도록 고침. 실제로 도는 중(analyzing)이면 그대로 막음. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""워크플로우 단계별 상태 관리를 위한 공통 유틸리티."""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any, Dict
|
|
|
|
import aiomysql
|
|
|
|
# `project_workflow_stages.message` 는 varchar(255) — 넘치면 UPDATE 자체가 실패한다.
|
|
_STAGE_MESSAGE_MAX_LENGTH = 200
|
|
|
|
STAGE_KEYS = [
|
|
"FILE_INPUT", # 0 (B03)
|
|
"PREPROCESS", # 1 (B04)
|
|
"PROFILE", # 2 (B05)
|
|
"SECTION", # 3 (B06)
|
|
"DESIGN_DETAIL", # 4 (B07)
|
|
"QUANTITY", # 5 (B08)
|
|
"ESTIMATION", # 6 (B09)
|
|
]
|
|
|
|
|
|
async def initialize_project_stages(cursor: aiomysql.DictCursor, project_id: str) -> None:
|
|
"""프로젝트 생성 시 7개의 단계를 NOT_STARTED 상태로 시드한다."""
|
|
for stage_no, stage_key in enumerate(STAGE_KEYS):
|
|
await cursor.execute(
|
|
"""
|
|
INSERT INTO project_workflow_stages (
|
|
project_id, stage_no, stage_key, state, progress_percent
|
|
)
|
|
VALUES (%s, %s, %s, 'NOT_STARTED', 0)
|
|
ON DUPLICATE KEY UPDATE
|
|
state = 'NOT_STARTED',
|
|
progress_percent = 0,
|
|
params = NULL,
|
|
message = NULL,
|
|
started_at = NULL,
|
|
completed_at = NULL
|
|
""",
|
|
(project_id, stage_no, stage_key),
|
|
)
|
|
|
|
|
|
async def start_stage(
|
|
cursor: aiomysql.DictCursor,
|
|
project_id: str,
|
|
stage_no: int,
|
|
params: Dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""단계를 시작하여 IN_PROGRESS 상태로 만들고, 이후 단계들을 STALE로 전환한다."""
|
|
now = datetime.utcnow()
|
|
params_json = json.dumps(params, ensure_ascii=False) if params is not None else None
|
|
|
|
# 해당 단계 시작
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE project_workflow_stages
|
|
SET state = 'IN_PROGRESS',
|
|
progress_percent = 0,
|
|
params = COALESCE(%s, params),
|
|
message = NULL,
|
|
started_at = %s,
|
|
completed_at = NULL
|
|
WHERE project_id = %s AND stage_no = %s
|
|
""",
|
|
(params_json, now, project_id, stage_no),
|
|
)
|
|
|
|
# 역방향 재작업 무효화 (stale 전파): 이후 단계 중 COMPLETE인 것들을 STALE로 변경
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE project_workflow_stages
|
|
SET state = 'STALE'
|
|
WHERE project_id = %s AND stage_no > %s AND state = 'COMPLETE'
|
|
""",
|
|
(project_id, stage_no),
|
|
)
|
|
|
|
# projects.status 캐시 업데이트 (호환성 유지)
|
|
status_str = f"WF{stage_no}_ANALYZING" if stage_no > 0 else "FILE_INPUT_PROCESSING"
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE projects
|
|
SET status = %s, updated_at = %s
|
|
WHERE id = %s
|
|
""",
|
|
(status_str, now, project_id),
|
|
)
|
|
|
|
|
|
async def complete_stage(cursor: aiomysql.DictCursor, project_id: str, stage_no: int) -> None:
|
|
"""단계를 완료 상태로 전환한다."""
|
|
now = datetime.utcnow()
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE project_workflow_stages
|
|
SET state = 'COMPLETE',
|
|
progress_percent = 100,
|
|
completed_at = %s
|
|
WHERE project_id = %s AND stage_no = %s
|
|
""",
|
|
(now, project_id, stage_no),
|
|
)
|
|
|
|
# projects.status 캐시 업데이트 (호환성 유지)
|
|
status_str = f"WF{stage_no}_COMPLETE" if stage_no > 0 else "FILE_UPLOADED"
|
|
if stage_no == 6:
|
|
status_str = "DONE"
|
|
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE projects
|
|
SET status = %s, updated_at = %s
|
|
WHERE id = %s
|
|
""",
|
|
(status_str, now, project_id),
|
|
)
|
|
|
|
|
|
async def update_stage_progress(
|
|
cursor: aiomysql.DictCursor, project_id: str, stage_no: int, progress: int
|
|
) -> None:
|
|
"""단계 진행률을 업데이트한다."""
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE project_workflow_stages
|
|
SET progress_percent = %s
|
|
WHERE project_id = %s AND stage_no = %s
|
|
""",
|
|
(progress, project_id, stage_no),
|
|
)
|
|
|
|
|
|
async def fail_stage(
|
|
cursor: aiomysql.DictCursor, project_id: str, stage_no: int, message: str
|
|
) -> None:
|
|
"""단계를 실패 상태로 전환한다.
|
|
|
|
`message`는 컬럼 길이에 맞춰 잘라 넣는다. 예외 문자열을 그대로 넣었다가 실패 기록
|
|
자체가 `DataError (1406, "Data too long for column 'message'")`로 죽어, 단계가 FAILED로
|
|
가지 못하고 화면이 영영 "분석 중"에 머문 사고가 있었다(2026-08-08 E2E 점검).
|
|
원문 전체는 호출부가 이미 로그에 남긴다.
|
|
"""
|
|
now = datetime.utcnow()
|
|
trimmed = (message or "").strip()
|
|
if len(trimmed) > _STAGE_MESSAGE_MAX_LENGTH:
|
|
trimmed = trimmed[: _STAGE_MESSAGE_MAX_LENGTH - 1] + "…"
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE project_workflow_stages
|
|
SET state = 'FAILED',
|
|
message = %s
|
|
WHERE project_id = %s AND stage_no = %s
|
|
""",
|
|
(trimmed, project_id, stage_no),
|
|
)
|
|
|
|
# projects.status 캐시 업데이트
|
|
status_str = f"WF{stage_no}_FAILED" if stage_no > 0 else "FILE_INPUT_FAILED"
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE projects
|
|
SET status = %s, updated_at = %s
|
|
WHERE id = %s
|
|
""",
|
|
(status_str, now, project_id),
|
|
)
|
|
|
|
|
|
async def get_workflow_state(cursor: aiomysql.DictCursor, project_id: str) -> Dict[str, Any]:
|
|
"""프로젝트의 모든 단계 상태를 조회하여 요약 및 배열로 반환한다.
|
|
|
|
프로젝트 이름도 함께 싣는다 (2026-09-04 사용자 지시) — B03~B08 좌측 제목 줄 오른쪽에
|
|
이름을 붙이는데, 화면이 들고 있는 것은 프로젝트 id 뿐이라 여기서 내려 준다. 새로고침·
|
|
주소 직접 입력으로 들어와도 같은 값이 따라온다.
|
|
"""
|
|
await cursor.execute("SELECT name FROM projects WHERE id = %s", (project_id,))
|
|
name_row = await cursor.fetchone()
|
|
project_name = name_row["name"] if name_row else None
|
|
|
|
await cursor.execute(
|
|
"""
|
|
SELECT stage_no, stage_key, state, progress_percent, params, message,
|
|
DATE_FORMAT(started_at, '%%Y-%%m-%%dT%%H:%%i:%%s') AS started_at,
|
|
DATE_FORMAT(completed_at, '%%Y-%%m-%%dT%%H:%%i:%%s') AS completed_at
|
|
FROM project_workflow_stages
|
|
WHERE project_id = %s
|
|
ORDER BY stage_no ASC
|
|
""",
|
|
(project_id,),
|
|
)
|
|
rows = await cursor.fetchall()
|
|
|
|
if not rows:
|
|
return {
|
|
"project_id": project_id,
|
|
"project_name": project_name,
|
|
"current_stage": 0,
|
|
"stages": [],
|
|
}
|
|
|
|
stages_list = []
|
|
for r in rows:
|
|
params_val = None
|
|
if r.get("params"):
|
|
try:
|
|
params_val = (
|
|
json.loads(r["params"]) if isinstance(r["params"], str) else r["params"]
|
|
)
|
|
except Exception:
|
|
params_val = r["params"]
|
|
|
|
stages_list.append(
|
|
{
|
|
"stage_no": r["stage_no"],
|
|
"stage_key": r["stage_key"],
|
|
"state": r["state"],
|
|
"progress_percent": r["progress_percent"],
|
|
"params": params_val,
|
|
"message": r["message"],
|
|
"started_at": r["started_at"],
|
|
"completed_at": r["completed_at"],
|
|
}
|
|
)
|
|
|
|
current_stage = 0
|
|
for stage in stages_list:
|
|
if stage["stage_no"] == 0:
|
|
continue
|
|
prev_stage = stages_list[stage["stage_no"] - 1]
|
|
if prev_stage["state"] == "COMPLETE":
|
|
current_stage = stage["stage_no"]
|
|
else:
|
|
break
|
|
|
|
return {
|
|
"project_id": project_id,
|
|
"project_name": project_name,
|
|
"current_stage": current_stage,
|
|
"stages": stages_list,
|
|
}
|
|
|
|
|
|
async def is_analysis_running(cursor: aiomysql.DictCursor, project_id: str) -> bool:
|
|
"""전처리(1단계)가 도는 중인지. 도는 동안 새 자료를 받으면 분석 2개가 같은 산출물
|
|
경로를 놓고 부딪혀 뒤에 시작한 쪽이 죽는다(2026-08-08 E2E 점검에서 실제 발생).
|
|
|
|
화면은 업로드 버튼을 잠가 이 상황을 막지만, 새로고침·다른 탭·보관함 연결로 우회할 수
|
|
있어 서버에서도 막는다.
|
|
"""
|
|
await cursor.execute(
|
|
"""
|
|
SELECT state
|
|
FROM project_workflow_stages
|
|
WHERE project_id = %s AND stage_no = 1
|
|
""",
|
|
(project_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
return False
|
|
state = row["state"] if isinstance(row, dict) else row[0]
|
|
if str(state) != "IN_PROGRESS":
|
|
return False
|
|
# 자동 확정이 보류되면 1단계는 사용자가 B04에서 모델을 고를 때까지 IN_PROGRESS 로
|
|
# 남는다 — 계산은 이미 끝난 상태다. 그 사이에도 새 자료는 받아야 한다
|
|
# (2026-09-04 사용자 지시 — 같은 프로젝트로 전처리를 되풀이 시험).
|
|
return not await _surface_run_settled(cursor, project_id)
|
|
|
|
|
|
# 계산이 끝나 사용자의 다음 조작을 기다리는 진행 단계 — 도는 중이 아니다.
|
|
_SETTLED_SURFACE_STAGES = frozenset({"awaiting_confirmation", "completed", "failed"})
|
|
|
|
|
|
async def _surface_run_settled(cursor: aiomysql.DictCursor, project_id: str) -> bool:
|
|
"""전처리 진행 파일이 「끝났음」으로 적혀 있는가. 파일이 없으면 판단하지 않는다."""
|
|
from pathlib import Path
|
|
|
|
from B04_PreProcess.B04_PreProcess_Router_Progress import read_surface_progress
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
|
|
await cursor.execute(
|
|
"SELECT storage_path FROM projects WHERE id = %s AND deleted_at IS NULL",
|
|
(project_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
return False
|
|
storage_path = row["storage_path"] if isinstance(row, dict) else row[0]
|
|
if not storage_path:
|
|
return False
|
|
try:
|
|
progress = read_surface_progress(Path(resolve_stored_project_path(str(storage_path))))
|
|
except (OSError, ValueError):
|
|
return False
|
|
stage = str((progress or {}).get("current_stage") or "")
|
|
return stage in _SETTLED_SURFACE_STAGES
|
|
|
|
|
|
async def reset_stages_after_input_change(cursor: aiomysql.DictCursor, project_id: str) -> None:
|
|
"""입력 자료가 갈렸을 때 1단계(전처리) 이후를 처음 상태로 되돌린다.
|
|
|
|
새 자료로 다시 계산해 덮어쓰므로, 옛 자료 기준의 진행 표시가 남아 있으면 사용자가 이미
|
|
끝난 단계로 착각한다. 파일 입력(0단계)은 방금 끝났으니 건드리지 않는다.
|
|
"""
|
|
await cursor.execute(
|
|
"""
|
|
UPDATE project_workflow_stages
|
|
SET state = 'NOT_STARTED',
|
|
progress_percent = 0,
|
|
params = NULL,
|
|
message = NULL,
|
|
started_at = NULL,
|
|
completed_at = NULL
|
|
WHERE project_id = %s AND stage_no >= 1
|
|
""",
|
|
(project_id,),
|
|
)
|