feat(dev): 확정 없이 다음으로 — 개발 전용 잠금 해제 (신규 파일만, 1/2)

공용 파일 두 걸음 규칙 — 이 커밋은 신규 파일뿐이고 main.py 등록은 다음 커밋. 먼저 push 해야 다른 창이 등록 줄만 받고 파일이 없어 죽는 일이 없음. 까닭 — 단계마다 [확정]을 해야 다음 페이지가 열려서, 상세 설계를 확정하기 전에는 B08·B09 를 볼 수 없음. 오늘 화면 검증이 그 자리에서 두 번 막혔음. ⚠ 계산을 대신 돌리지 않음 — 잠금만 품 - 바꾸는 것은 project_workflow_stages.state 한 칸뿐 - 확정은 전 측점을 다시 계산해 정본에 쓰는 것이라, 흉내 내면 「확정 안 했는데 확정된 값」이 생겨 막힌 것보다 나쁨 - 값이 없으면 B08 이 「미확보」로 뜨는 것이 정상 ⚠ 문은 서버가 정본 — ENVIRONMENT 가 개발이 아니면 세 입구 모두 403 (화면에서 단추를 숨겨도 API 가 열려 있으면 소용없음) 되돌리기 — 푼 단계의 이전 상태를 message 에 DEV_UNLOCK:<옛상태> 로 적어 두고 DELETE 가 그대로 복원. 그 표시가 없는 단계(진짜 확정)는 안 건드림. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 06:39:05 +09:00
parent de31cd9c42
commit 618b6cf4c5
2 changed files with 278 additions and 0 deletions
@@ -0,0 +1,165 @@
+"""개발환경 전용 — **확정을 거치지 않고 다음 단계로 넘어가게** 하는 자리.
+
+왜 있나
+    프로그램은 단계마다 [확정]을 해야 다음 페이지가 열린다. 그래서 **상세 설계를 확정하기
+    전에는 B08·B09 를 아예 볼 수 없다.** 화면 검증을 하려면 매번 남의 프로젝트 확정 상태에
+    매달려야 했다(2026-09-08 실제로 그 자리에서 두 창이 막혔다).
+
+⚠⚠ **계산을 대신 돌리지 않는다 — 잠금만 푼다.**
+    [확정]은 **전 측점을 다시 계산해 정본에 쓰는 것**이다. 여기서 그 계산을 흉내 내면
+    「확정 안 했는데 확정된 값」이 생겨 **막힌 것보다 더 나쁘다.** 그래서 이 모듈이 바꾸는
+    것은 `project_workflow_stages.state` **한 칸뿐**이다. 값이 없으면 B08 이
+    「미확보」로 뜨는 것이 **정상이고 그것이 옳은 화면**이다.
+
+⚠ **문은 서버가 정본이다.**
+    화면에서 단추를 숨기는 것만으로는 API 가 그대로 뚫려 있다. 그래서 이 모듈이
+    `ENVIRONMENT` 를 보고 **운영에서는 아예 거절한다.** 화면 쪽 `import.meta.env.DEV` 는
+    보조일 뿐이다.
+
+⚠ **되돌릴 수 있어야 한다.**
+    푼 단계와 그 **이전 상태**를 `dev_unlock` 칸에 적어 두고, 되돌리기가 그대로 복원한다.
+    안 그러면 검증용 프로젝트가 이상한 상태로 굳는다.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from typing import Any
+
+import aiomysql
+
+from common_util.common_util_workflow_state import STAGE_KEYS
+from config.config_system import ENVIRONMENT
+
+#: 개발환경으로 보는 값. 그 밖(staging·production)에서는 이 기능이 아예 안 돈다.
+DEV_ENVIRONMENTS = frozenset({"development", "dev", "local", "test"})
+
+#: 되돌리기용 기록을 남기는 자리. 프로젝트 저장 폴더가 아니라 **DB 안**에 둔다 —
+#: 상태와 같은 곳에 있어야 둘이 어긋나지 않는다.
+UNLOCK_MESSAGE_PREFIX = "DEV_UNLOCK:"
+
+
+class DevUnlockDisabled(RuntimeError):
+    """개발환경이 아니어서 거절함. **이 예외가 곧 운영 쪽 문**이다."""
+
+
+def is_dev_environment() -> bool:
+    """지금 환경에서 이 기능을 켜도 되는가."""
+    return str(ENVIRONMENT or "").strip().lower() in DEV_ENVIRONMENTS
+
+
+def require_dev_environment() -> None:
+    """개발환경이 아니면 **여기서 멈춘다.** 화면이 아니라 서버가 막는 자리다."""
+    if not is_dev_environment():
+        raise DevUnlockDisabled(f"개발환경에서만 쓸 수 있습니다 (지금 환경: {ENVIRONMENT}).")
+
+
+async def unlock_stages(
+    cursor: aiomysql.DictCursor, project_id: str, up_to_stage: int
+) -> dict[str, Any]:
+    """`up_to_stage` 까지의 단계를 **상태만** COMPLETE 로 만든다.
+
+    ⚠ 계산·저장은 하지 않는다. 되돌릴 수 있게 **이전 상태를 함께 적어 둔다.**
+    이미 COMPLETE 인 단계는 건드리지 않는다 — 되돌릴 때 남의 확정까지 풀면 안 된다.
+    """
+    require_dev_environment()
+    if not 0 <= up_to_stage < len(STAGE_KEYS):
+        raise ValueError(f"단계 번호가 범위를 벗어났습니다: {up_to_stage}")
+
+    await cursor.execute(
+        """
+        SELECT stage_no, state, message
+        FROM project_workflow_stages
+        WHERE project_id = %s AND stage_no <= %s
+        ORDER BY stage_no ASC
+        """,
+        (project_id, up_to_stage),
+    )
+    rows = await cursor.fetchall()
+
+    changed: list[dict[str, Any]] = []
+    now = datetime.utcnow()
+    for row in rows:
+        if row["state"] == "COMPLETE":
+            continue  # 진짜로 확정된 단계 — 손대지 않는다.
+        changed.append({"stage_no": int(row["stage_no"]), "state": row["state"]})
+        note = f"{UNLOCK_MESSAGE_PREFIX}{row['state']}"
+        await cursor.execute(
+            """
+            UPDATE project_workflow_stages
+            SET state = 'COMPLETE',
+                progress_percent = 100,
+                completed_at = %s,
+                message = %s
+            WHERE project_id = %s AND stage_no = %s
+            """,
+            (now, note, project_id, int(row["stage_no"])),
+        )
+    return {
+        "unlocked": changed,
+        "up_to_stage": up_to_stage,
+        "note": (
+            "확정을 건너뛰고 잠금만 풀었습니다 — 계산은 돌지 않았습니다. "
+            "값이 비어 보이는 것은 정상입니다."
+        ),
+    }
+
+
+async def relock_stages(cursor: aiomysql.DictCursor, project_id: str) -> dict[str, Any]:
+    """우회로 푼 단계를 **원래 상태로 되돌린다.**
+
+    ⚠ `DEV_UNLOCK:` 표시가 붙은 단계만 되돌린다 — 그 표시가 없으면 **진짜 확정**이라
+    건드리면 안 된다. 표시 뒤에 적어 둔 옛 상태를 그대로 복원한다.
+    """
+    require_dev_environment()
+    await cursor.execute(
+        """
+        SELECT stage_no, message
+        FROM project_workflow_stages
+        WHERE project_id = %s AND message LIKE %s
+        ORDER BY stage_no ASC
+        """,
+        (project_id, f"{UNLOCK_MESSAGE_PREFIX}%"),
+    )
+    rows = await cursor.fetchall()
+
+    restored: list[dict[str, Any]] = []
+    for row in rows:
+        previous = str(row["message"])[len(UNLOCK_MESSAGE_PREFIX) :].strip() or "NOT_STARTED"
+        restored.append({"stage_no": int(row["stage_no"]), "state": previous})
+        await cursor.execute(
+            """
+            UPDATE project_workflow_stages
+            SET state = %s,
+                progress_percent = 0,
+                completed_at = NULL,
+                message = NULL
+            WHERE project_id = %s AND stage_no = %s
+            """,
+            (previous, project_id, int(row["stage_no"])),
+        )
+    return {"relocked": restored}
+
+
+async def unlock_status(cursor: aiomysql.DictCursor, project_id: str) -> dict[str, Any]:
+    """지금 우회로 열려 있는 단계 목록. **화면이 띄울 안내의 근거**다."""
+    await cursor.execute(
+        """
+        SELECT stage_no, message
+        FROM project_workflow_stages
+        WHERE project_id = %s AND message LIKE %s
+        ORDER BY stage_no ASC
+        """,
+        (project_id, f"{UNLOCK_MESSAGE_PREFIX}%"),
+    )
+    rows = await cursor.fetchall()
+    return {
+        "dev_environment": is_dev_environment(),
+        "bypassed_stages": [int(row["stage_no"]) for row in rows],
+    }
+
+
+def as_json(payload: dict[str, Any]) -> str:
+    """로그용 — 한글이 깨지지 않게."""
+    return json.dumps(payload, ensure_ascii=False)
@@ -0,0 +1,113 @@
+"""개발환경 전용 — 「확정 없이 다음으로」 API.
+
+⚠ **문은 여기가 정본이다.** 화면에서 단추를 숨겨도 API 가 열려 있으면 아무 소용이 없다.
+   그래서 세 입구 모두 `require_dev_environment()` 를 먼저 부르고, 운영에서는 **403** 으로
+   거절한다. 프론트의 `import.meta.env.DEV` 는 보조일 뿐이다.
+
+⚠ **계산을 대신 돌리지 않는다.** 여기서 바뀌는 것은 `project_workflow_stages.state`
+   한 칸뿐이다. 자세한 까닭은 `common_util_dev_unlock` 의 머리말을 볼 것.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import UUID
+
+import aiomysql
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common_util.common_util_dev_unlock import (
+    DevUnlockDisabled,
+    relock_stages,
+    unlock_stages,
+    unlock_status,
+)
+from config.config_db import run_with_connection
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/projects", tags=["DEV Unlock"])
+
+#: 기본은 B08(수량산출)까지 — 검증이 가장 자주 막히던 자리다.
+DEFAULT_UP_TO_STAGE = 5
+
+
+class UnlockBody(BaseModel):
+    """어디까지 열 것인가. 안 주면 B08 까지."""
+
+    up_to_stage: int = DEFAULT_UP_TO_STAGE
+
+
+async def _with_cursor(connection: aiomysql.Connection, call: Any, *args: Any) -> Any:
+    """쓰기라 **한 커넥션·한 트랜잭션**으로 묶는다(`run_with_connection` 주석 참조)."""
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        result = await call(cursor, *args)
+    await connection.commit()
+    return result
+
+
+@router.get("/{project_id}/dev/unlock")
+async def get_unlock_status(project_id: UUID) -> JSONResponse:
+    """지금 우회로 열려 있는 단계. **화면 안내의 근거**다."""
+
+    async def call(connection: aiomysql.Connection) -> dict[str, Any]:
+        async with connection.cursor(aiomysql.DictCursor) as cursor:
+            return await unlock_status(cursor, str(project_id))
+
+    try:
+        payload = await run_with_connection(call)
+    except Exception:
+        logger.exception("개발 우회 상태 조회 실패: project_id=%s", project_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "우회 상태를 읽지 못했습니다."},
+        )
+    return JSONResponse(content={"status": "success", **payload})
+
+
+@router.post("/{project_id}/dev/unlock")
+async def post_unlock(project_id: UUID, body: UnlockBody | None = None) -> JSONResponse:
+    """확정을 건너뛰고 **잠금만** 푼다. 개발환경이 아니면 403."""
+    up_to = (body or UnlockBody()).up_to_stage
+
+    async def call(connection: aiomysql.Connection) -> dict[str, Any]:
+        return await _with_cursor(connection, unlock_stages, str(project_id), up_to)
+
+    try:
+        payload = await run_with_connection(call)
+    except DevUnlockDisabled as error:
+        return JSONResponse(status_code=403, content={"status": "error", "message": str(error)})
+    except ValueError as error:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
+    except Exception:
+        logger.exception("개발 우회 실패: project_id=%s", project_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "단계를 열지 못했습니다."},
+        )
+    logger.info("DEV 우회: project_id=%s 까지=%s", project_id, up_to)
+    return JSONResponse(content={"status": "success", **payload})
+
+
+@router.delete("/{project_id}/dev/unlock")
+async def delete_unlock(project_id: UUID) -> JSONResponse:
+    """우회로 연 단계를 **원래 상태로 되돌린다.** 진짜 확정은 안 건드린다."""
+
+    async def call(connection: aiomysql.Connection) -> dict[str, Any]:
+        return await _with_cursor(connection, relock_stages, str(project_id))
+
+    try:
+        payload = await run_with_connection(call)
+    except DevUnlockDisabled as error:
+        return JSONResponse(status_code=403, content={"status": "error", "message": str(error)})
+    except Exception:
+        logger.exception("개발 우회 되돌리기 실패: project_id=%s", project_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "단계를 되돌리지 못했습니다."},
+        )
+    logger.info("DEV 우회 되돌림: project_id=%s", project_id)
+    return JSONResponse(content={"status": "success", **payload})