From 618b6cf4c58098172db9be5980f92f1b82aad645 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 8 Sep 2026 06:39:05 +0900 Subject: [PATCH] =?UTF-8?q?feat(dev):=20=ED=99=95=EC=A0=95=20=EC=97=86?= =?UTF-8?q?=EC=9D=B4=20=EB=8B=A4=EC=9D=8C=EC=9C=BC=EB=A1=9C=20=E2=80=94=20?= =?UTF-8?q?=EA=B0=9C=EB=B0=9C=20=EC=A0=84=EC=9A=A9=20=EC=9E=A0=EA=B8=88=20?= =?UTF-8?q?=ED=95=B4=EC=A0=9C=20(=EC=8B=A0=EA=B7=9C=20=ED=8C=8C=EC=9D=BC?= =?UTF-8?q?=EB=A7=8C,=201/2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 공용 파일 두 걸음 규칙 — 이 커밋은 신규 파일뿐이고 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) --- common_util/common_util_dev_unlock.py | 165 +++++++++++++++++++ common_util/common_util_dev_unlock_router.py | 113 +++++++++++++ 2 files changed, 278 insertions(+) create mode 100644 common_util/common_util_dev_unlock.py create mode 100644 common_util/common_util_dev_unlock_router.py diff --git a/common_util/common_util_dev_unlock.py b/common_util/common_util_dev_unlock.py new file mode 100644 index 00000000..41736b81 --- /dev/null +++ b/common_util/common_util_dev_unlock.py @@ -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) diff --git a/common_util/common_util_dev_unlock_router.py b/common_util/common_util_dev_unlock_router.py new file mode 100644 index 00000000..852301d5 --- /dev/null +++ b/common_util/common_util_dev_unlock_router.py @@ -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})