eomsangdonandClaude Opus 5
018ebe7f40
계획서 0-8 의 남은 하나. 서버는 분석이 도는 동안 새 자료를 이미 막고 있었고 (`is_analysis_running`), 없던 것은 **막힌 까닭을 사람에게 보이는 것**이었음. 여럿이 한 프로젝트를 보면 「왜 안 올라가지」가 됨. - 분석을 띄우는 **경로 넷 모두** 시작한 사람을 `params.started_by` 로 담음. - `analysis_lock_owner` 가 그 id 로 이름을 붙여 냄 — 옛 자료는 id 가 없어 **이름 없이 잠금만** 돎(막는 것이 먼저). - `/upload-overview` 응답에 `analysis_lock` 을 실어 B03 화면이 띠로 보임. 화면 실측 — 「엄상돈 님이 올린 자료를 분석하는 중입니다」 띠가 경고색으로 뜸(응답을 잠깐 가로채 확인하고 곧바로 되돌림). 지금 도는 분석이 없어 실제 잠금 화면은 다음 업로드 때 볼 것. 시험 넷 추가(도는 중 아님·이름 붙음·옛 자료·경로 넷 대조), 전체 1211 통과·실패 0. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
370 lines
13 KiB
Python
370 lines
13 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)
|
|
|
|
|
|
async def analysis_lock_owner(
|
|
cursor: aiomysql.DictCursor, project_id: str
|
|
) -> Dict[str, Any] | None:
|
|
"""**지금 누가 올리는 중인지** — 도는 중이 아니면 `None`.
|
|
|
|
잠금 자체는 `is_analysis_running` 이 판정한다. 여기서는 그 잠금에 **이름을 붙인다** —
|
|
한 프로젝트를 여럿이 볼 때 「왜 못 올리지」가 아니라 「누가 올리는 중이구나」가 되게.
|
|
시작한 사람은 `start_stage` 가 담아 둔 `params.started_by`(사용자 id)에서 읽고,
|
|
이름은 그때그때 `users` 에서 가져온다(옛 자료는 id 가 없어 이름 없이 잠금만 돈다).
|
|
"""
|
|
if not await is_analysis_running(cursor, project_id):
|
|
return None
|
|
await cursor.execute(
|
|
"""
|
|
SELECT params, started_at
|
|
FROM project_workflow_stages
|
|
WHERE project_id = %s AND stage_no = 1
|
|
""",
|
|
(project_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
return None
|
|
raw = row["params"] if isinstance(row, dict) else row[0]
|
|
started_at = row["started_at"] if isinstance(row, dict) else row[1]
|
|
params: Dict[str, Any] = {}
|
|
if isinstance(raw, str):
|
|
try:
|
|
params = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
params = {}
|
|
elif isinstance(raw, dict):
|
|
params = raw
|
|
user_id = params.get("started_by")
|
|
name: str | None = None
|
|
email: str | None = None
|
|
if user_id is not None:
|
|
await cursor.execute("SELECT name, email FROM users WHERE id = %s", (user_id,))
|
|
user = await cursor.fetchone()
|
|
if user:
|
|
name = user["name"] if isinstance(user, dict) else user[0]
|
|
email = user["email"] if isinstance(user, dict) else user[1]
|
|
return {
|
|
"running": True,
|
|
"user_id": user_id,
|
|
"name": name,
|
|
"email": email,
|
|
"started_at": started_at.isoformat() if started_at else None,
|
|
}
|
|
|
|
|
|
# 계산이 끝나 사용자의 다음 조작을 기다리는 진행 단계 — 도는 중이 아니다.
|
|
_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,),
|
|
)
|