Aislo/common_util/common_util_dev_unlock.py

"""개발환경 전용 — **확정을 거치지 않고 다음 단계로 넘어가게** 하는 자리.

왜 있나
    프로그램은 단계마다 [확정]을 해야 다음 페이지가 열린다. 그래서 **상세 설계를 확정하기
    전에는 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)