Aislo/B07_DesignDetail/B07_DesignDetail_Router.py

"""B06 확정 종·횡단 산출물을 B07 CAD 도면으로 변환하는 라우터."""

import asyncio
import logging
import re
from base64 import b64encode
from pathlib import Path, PurePosixPath
from typing import Any
from uuid import UUID

from aiomysql import DictCursor
from fastapi import APIRouter
from fastapi.responses import JSONResponse

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B06_Section.B06_Section_Repository import (
    get_confirmed_route_context,
    get_cross_section_design,
    get_cross_section_designs,
    get_longitudinal_section,
    get_project_standard_cross_section,
    merge_cross_section_design_by_round,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Sheet import (
    CROSS_SHEET_ID,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Table import (
    extract_quantity_table,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Template import (
    use_company_templates,
    use_title_fields,
)
from B07_DesignDetail.B07_DesignDetail_Router_Standard import section_modes_of
from B07_DesignDetail.B07_DesignDetail_Router_Support import (
    CROSS_STANDARD_ID,
    LANDUSE_ID,
    LIDAR_ID,
    MASS_HAUL_ID,
    PLAN_ID,
    WATERSHED_ID,
    _cross_sheet_plan,
    _drawing_list,
    _invalidate_drawing,
    _read_drawing,
    _read_json,
    _recompute_confirmed_design,
    _store_confirmed_drawing,
    landuse_source,
    lidar_source,
    plan_source,
    watershed_source,
)
from B07_DesignDetail.B07_DesignDetail_Schema import (
    DesignDrawingConfirmRequest,
    DesignDrawingConfirmResponse,
    DesignDrawingInvalidateResponse,
    DesignDrawingListResponse,
    DesignDrawingResponse,
)
from common_util.common_util_dev_unlock import is_dev_environment, unlock_status
from common_util.common_util_drainage_context import load_drainage_context
from common_util.common_util_storage import read_stored_asset, resolve_stored_project_path
from common_util.common_util_workflow_state import complete_stage, start_stage
from config.config_db import get_db_pool

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B07 Design Detail"])

_CROSS_ID = re.compile(r"^cross_(\d+)m$")
_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")


#: B07(상세설계) 워크플로 단계 번호 — 개발 우회로가 이 단계를 풀었는지 본다.
_B07_STAGE_NO = 4


async def _confirmed_source(project_id: UUID) -> tuple[int, Path, Path, bool]:
    """확정된 B06 종단 레코드와 프로젝트 저장 경로. 넷째는 **개발 우회로로 열렸나**.

    ⚠ **문은 서버가 막는다.** 화면 표시로 여는 것이 아니라 여기서 막고, **운영에서는 그대로
      409** 다(`is_dev_environment()` 가 거짓이면 우회가 아예 없다).
    ⚠ **개발환경이라고 무조건 열지 않는다** — `dev/unlock` 표식이 실제로 있을 때만 연다.
      그래야 되돌리기(DELETE)로 **다시 닫힌다**.
    ⚠ **레코드는 손대지 않는다** — `longitudinal_sections.status` 는 `DRAFT` 그대로 두고
      **읽는 쪽만** 우회한다. 상태를 올리면 그것이 곧 「확정 흉내」라 `common_util_dev_unlock`
      이 스스로 금지한 자리와 같아진다.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        route_context = await get_confirmed_route_context(connection, project_id)
        if not route_context:
            raise FileNotFoundError("확정된 경로가 없습니다.")
        route_id = int(route_context["route_id"])
        longitudinal = await get_longitudinal_section(connection, project_id, route_id)
        bypassed = False
        if not longitudinal or longitudinal.get("status") != "CONFIRMED":
            if is_dev_environment():
                async with connection.cursor(DictCursor) as cursor:
                    status = await unlock_status(cursor, str(project_id))
                bypassed = _B07_STAGE_NO in (status.get("bypassed_stages") or [])
            if not (bypassed and longitudinal):
                raise PermissionError("B06 종·횡단 확정 후 상세 설계를 진행할 수 있습니다.")
            logger.info("B07 개발 우회로로 열림 — 확정 건너뜀: project_id=%s", project_id)
        stored_path = await get_project_storage_relative_path(connection, project_id)

    root = Path(resolve_stored_project_path(stored_path)).resolve()
    longitudinal_path = (root / str(longitudinal["longitudinal_file_path"])).resolve()
    if root not in longitudinal_path.parents or not longitudinal_path.is_file():
        raise FileNotFoundError("B06 종단면 파일을 찾을 수 없습니다.")
    return route_id, root, longitudinal_path, bypassed


async def _company_dir(project_id: UUID) -> Path:
    """프로젝트 저장 경로에서 회사 폴더를 얻는다 — `storage/{회사}/{사용자}/{프로젝트}`.

    회사 도각은 회사 폴더에 산다(로고·서명과 같은 결). DB를 한 번 더 뒤지지 않는다.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored_path = await get_project_storage_relative_path(connection, project_id)
    root = Path(resolve_stored_project_path(stored_path)).resolve()
    return root.parent.parent


def _asset_data_url(relative_path: str | None) -> str:
    """회사 로고·개인 서명 파일을 CAD `ImageEntity`가 읽는 data URL로 만든다.

    파일이 없거나 경로가 수상하면 빈 문자열 — 그 자리는 그림째 빠진다
    (`_fill_placeholders`). 그림은 DB에 경로만 담는 기존 방식 그대로다.
    """
    blob = read_stored_asset(relative_path)
    if not blob:
        return ""
    suffix = PurePosixPath(str(relative_path)).suffix.lower()
    mime = "image/svg+xml" if suffix == ".svg" else f"image/{suffix.lstrip('.') or 'png'}"
    return f"data:{mime};base64,{b64encode(blob).decode('ascii')}"


async def title_block_fields(project_id: UUID) -> dict[str, str]:
    """도각 표제란에 채울 값. **DB가 아는 것만** 담고 나머지는 담지 않는다.

    담지 않은 자리는 `_fill_placeholders`가 빈칸으로 지운다 — 도각 원본에 남의 값이
    박혀 있어도 도면에는 나가지 않는다(2026-08-31 사용자 확정, 이것이 1순위 목적).

    사람 배정(과업책임자·분야별책임자·설계자)은 `projects`의 FK를 따라간다. 설계자는
    배정이 없으면 프로젝트 소유자로 떨어진다 — 혼자 쓰는 계정에서도 칸이 차게.
    **이름이 들어가는 자리는 서명도 받는다**(2026-09-02 사용자 확정) — 세 사람 모두
    자기 계정의 `SIGNATURE` 자산을 따라가고, 안 올린 사람 자리는 그림째 빠진다.
    로고·서명은 회사 공유 자산(`company_assets`, 013)에서 오되 **원천이 다르다**
    (2026-09-02 사용자 확정, B01 인계). 서명은 프로젝트가 아니라 **사람 계정**에 붙으므로
    설계자(`designer_user_id`, 없으면 소유자) 계정의 `SIGNATURE` 자산을 따라간다. 로고는
    프로젝트가 고른 것이 우선이고, 안 골랐으면 **회사 등록 단계에서 받은 회사 로고**
    (`companies.logo_asset_id`, 014)로 떨어진다. 둘 다 없거나 자산이 지워졌으면 그림째
    빠진다. `projects.signature_asset_id` 는 컬럼만 남기고 더 읽지 않는다(B01 이 항상
    `null` 로 저장). 011 의 `companies.logo_path` · `users.signature_path` 도 안 읽는다.

    아직 못 채우는 자리와 이유:
      - 사업량·연도기번은 사람이 넣는 값이다(B01 프로젝트 수정 화면). 비어 있으면 빈칸.
      - 설계일자·도면번호는 여기서 담지 않는다 — 설계일자는 사용자가 넣은 `design_date`
        를 그대로 쓰고(확정일 자동이 아니다), 도면번호는 도면을 읽는 쪽이 manifest 에
        적힌 목록 순번을 얹는다.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT p.name, p.region, p.client_org, p.project_number, p.work_amount,
                   p.design_date, c.name, logo.file_path,
                   COALESCE(designer.name, owner.name), sig.file_path,
                   pm.name, lead.name, pm_sig.file_path, lead_sig.file_path
            FROM projects p
            LEFT JOIN companies c ON c.id = p.company_id
            LEFT JOIN users owner ON owner.id = p.user_id
            LEFT JOIN users designer ON designer.id = p.designer_user_id
            LEFT JOIN users pm ON pm.id = p.pm_user_id
            LEFT JOIN users lead ON lead.id = p.field_lead_user_id
            LEFT JOIN company_assets logo
                   ON logo.id = COALESCE(p.logo_asset_id, c.logo_asset_id)
                  AND logo.deleted_at IS NULL
            LEFT JOIN company_assets sig
                   ON sig.id = (
                        SELECT s.id
                          FROM company_assets s
                         WHERE s.user_id = COALESCE(p.designer_user_id, p.user_id)
                           AND s.kind = 'SIGNATURE'
                           AND s.deleted_at IS NULL
                         ORDER BY s.id DESC
                         LIMIT 1
                      )
            LEFT JOIN company_assets pm_sig
                   ON pm_sig.id = (
                        SELECT s.id
                          FROM company_assets s
                         WHERE s.user_id = p.pm_user_id
                           AND s.kind = 'SIGNATURE'
                           AND s.deleted_at IS NULL
                         ORDER BY s.id DESC
                         LIMIT 1
                      )
            LEFT JOIN company_assets lead_sig
                   ON lead_sig.id = (
                        SELECT s.id
                          FROM company_assets s
                         WHERE s.user_id = p.field_lead_user_id
                           AND s.kind = 'SIGNATURE'
                           AND s.deleted_at IS NULL
                         ORDER BY s.id DESC
                         LIMIT 1
                      )
            WHERE p.id = %s AND p.deleted_at IS NULL
            """,
            (str(project_id),),
        )
        row = await cursor.fetchone()
    if not row:
        return {}
    (
        name,
        region,
        client_org,
        project_number,
        work_amount,
        design_date,
        company,
        logo_path,
        designer,
        signature_path,
        pm,
        lead,
        pm_signature_path,
        lead_signature_path,
    ) = row
    fields = {
        "공사명": name,
        "위치": region,
        "시행청": client_org,
        "연도기번": project_number,
        "사업량": work_amount,
        # 도면 표기 관행대로 "2026. 09. 02." 꼴로 적는다.
        "설계일자": f"{design_date:%Y. %m. %d.}" if design_date else None,
        "용역회사": company,
        "설계자": designer,
        "과업책임자": pm,
        "분야별책임자": lead,
        "회사로고": _asset_data_url(logo_path),
        "설계자서명": _asset_data_url(signature_path),
        "과업책임자서명": _asset_data_url(pm_signature_path),
        "분야별책임자서명": _asset_data_url(lead_signature_path),
    }
    return {key: str(value) for key, value in fields.items() if value}


async def _designs_by_chainage(route_id: int) -> dict[int, dict[str, Any]]:
    """노선 전체의 측점별 설계 지정 {측점(m): design}. 장 배치·목록이 함께 쓴다."""
    pool = get_db_pool()
    async with pool.acquire() as connection:
        rows = await get_cross_section_designs(connection, route_id)
    return {
        int(round(float(row["chainage_m"]))): row["design"]
        for row in rows
        if isinstance(row, dict) and isinstance(row.get("design"), dict)
    }


@router.get("/{project_id}/design-drawings", response_model=DesignDrawingListResponse)
async def get_design_drawing_list(
    project_id: UUID,
) -> DesignDrawingListResponse | JSONResponse:
    """B07 좌측 패널용 도면 메타데이터만 캐시한다."""
    try:
        route_id, project_root, longitudinal_path, bypass = await _confirmed_source(project_id)
        designs = await _designs_by_chainage(route_id)
        modes = await section_modes_of(route_id)
        drawings = await asyncio.to_thread(
            _drawing_list, project_root, longitudinal_path, designs, modes
        )
        return DesignDrawingListResponse(
            project_id=str(project_id), route_id=route_id, drawings=drawings, dev_bypass=bypass
        )
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except PermissionError as exc:
        return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception("B07 도면 목록 조회 실패: project_id=%s", project_id)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "상세 설계 도면 목록을 읽지 못했습니다."},
        )


@router.get("/{project_id}/design-drawings/{drawing_id}", response_model=DesignDrawingResponse)
async def get_design_drawing(
    project_id: UUID, drawing_id: str
) -> DesignDrawingResponse | JSONResponse:
    """선택한 도면 원본 한 건만 읽어 CAD 스키마로 변환한다."""
    try:
        route_id, project_root, longitudinal_path, bypass = await _confirmed_source(project_id)
        # 이 회사가 고친 도각이 있으면 그것으로 그린다(없으면 프로그램 기본 도각).
        # 저장 경로는 `storage/{회사}/{사용자}/{프로젝트}` 이므로 두 단계 위가 회사 폴더다.
        use_company_templates(project_root.parent.parent)
        # 표제란 값도 같은 요청 문맥에 세운다 — 값이 없는 칸은 빈칸으로 나간다.
        use_title_fields(await title_block_fields(project_id))
        # 횡단도는 B06 지정 설계를 먼저 읽어 CAD 계획선(design_line)과 응답에 함께 쓴다.
        design: dict[str, Any] | None = None
        source_design: Any = None
        cross_match = _CROSS_ID.fullmatch(drawing_id)
        if cross_match:
            pool = get_db_pool()
            async with pool.acquire() as connection:
                design = await get_cross_section_design(
                    connection, route_id, int(cross_match.group(1))
                )
            source_design = design
        elif CROSS_SHEET_ID.fullmatch(drawing_id):
            # 장은 여러 측점을 담으므로 노선 전체 지정을 한 번에 읽어 넘긴다.
            source_design = await _designs_by_chainage(route_id)
        elif drawing_id == MASS_HAUL_ID:
            # 유토곡선은 B06 확정 시 종단 레코드에 저장해 둔 산출물을 그대로 쓴다.
            pool = get_db_pool()
            async with pool.acquire() as connection:
                section = await get_longitudinal_section(connection, project_id, route_id)
            source_design = ((section or {}).get("data") or {}).get("mass_haul")
        elif drawing_id == WATERSHED_ID:
            # 유역도는 B04 배수유역 산출물 + 도엽 배경을 사업지 CRS로 모아 넘긴다.
            context, reason = await load_drainage_context(project_id)
            if context is None:
                return JSONResponse(status_code=404, content={"status": "error", "message": reason})
            source_design = await asyncio.to_thread(watershed_source, context)
        elif drawing_id == CROSS_STANDARD_ID:
            # 표준 횡단면도는 B06 「표준 횡단면 설정」 저장값으로 그린다(없으면 config 기본값).
            pool = get_db_pool()
            async with pool.acquire() as connection:
                async with connection.cursor() as cursor:
                    await cursor.execute(
                        "SELECT company_id FROM projects WHERE id = %s AND deleted_at IS NULL",
                        (str(project_id),),
                    )
                    row = await cursor.fetchone()
                source_design = (
                    await get_project_standard_cross_section(connection, int(row[0]), project_id)
                    if row
                    else None
                )
        elif LIDAR_ID.fullmatch(drawing_id):
            # 라이다 계획평면도는 확정 DTM 격자로 음영기복 그림을 만들어 넘긴다.
            context, reason = await load_drainage_context(project_id)
            if context is None:
                return JSONResponse(status_code=404, content={"status": "error", "message": reason})
            longitudinal = await asyncio.to_thread(_read_json, longitudinal_path)
            source_design = await asyncio.to_thread(lidar_source, context, longitudinal, drawing_id)
        elif LANDUSE_ID.fullmatch(drawing_id):
            # 용지도도 같은 배경 창구를 쓴다 — 지적·행정 경계만 따로 읽는다.
            context, reason = await load_drainage_context(project_id)
            if context is None:
                return JSONResponse(status_code=404, content={"status": "error", "message": reason})
            longitudinal = await asyncio.to_thread(_read_json, longitudinal_path)
            source_design = await asyncio.to_thread(
                landuse_source, context, longitudinal, drawing_id
            )
        elif PLAN_ID.fullmatch(drawing_id):
            # 계획평면도는 유역도와 **같은 배경 창구**를 쓴다 — 자료 읽기·환산이 캐시된다.
            context, reason = await load_drainage_context(project_id)
            if context is None:
                return JSONResponse(status_code=404, content={"status": "error", "message": reason})
            longitudinal = await asyncio.to_thread(_read_json, longitudinal_path)
            source_design = await asyncio.to_thread(plan_source, context, longitudinal, drawing_id)
        kind, label, drawing, confirmed, quantity_table = await asyncio.to_thread(
            _read_drawing,
            project_root,
            longitudinal_path,
            drawing_id,
            source_design,
            await section_modes_of(route_id),
        )
        return DesignDrawingResponse(
            project_id=str(project_id),
            route_id=route_id,
            id=drawing_id,
            kind=kind,
            label=label,
            drawing=drawing,
            confirmed=confirmed,
            quantity_table=quantity_table,
            design=design,
        )
    except ValueError as exc:
        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except PermissionError as exc:
        return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception(
            "B07 단건 도면 조회 실패: project_id=%s drawing_id=%s", project_id, drawing_id
        )
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "상세 설계 도면을 읽지 못했습니다."},
        )


@router.put(
    "/{project_id}/design-drawings/{drawing_id}/confirm",
    response_model=DesignDrawingConfirmResponse,
)
async def confirm_design_drawing(
    project_id: UUID, drawing_id: str, request: DesignDrawingConfirmRequest
) -> DesignDrawingConfirmResponse | JSONResponse:
    """현재 편집 도면을 영구 저장하고 도면별 확정 상태를 기록한다.

    횡단도 확정 시 B06 지정 잠정치를 동일 엔진으로 재계산해 확정치로 승격·저장한다.
    """
    try:
        route_id, project_root, longitudinal_path, bypass = await _confirmed_source(project_id)
        designs = await _designs_by_chainage(route_id)
        items = await asyncio.to_thread(_drawing_list, project_root, longitudinal_path, designs)
        item = next((candidate for candidate in items if candidate.id == drawing_id), None)
        if not item:
            raise FileNotFoundError("확정할 도면을 찾을 수 없습니다.")
        sheet = None
        if CROSS_SHEET_ID.fullmatch(drawing_id):
            plan = await asyncio.to_thread(
                _cross_sheet_plan, project_root, longitudinal_path, designs
            )
            sheet = next((entry for entry in plan if entry["id"] == drawing_id), None)
        # 수량표는 CAD 테이블(Text 엔티티)에서 역추출을 우선하고, 없으면 요청 본문 폴백.
        # 장에는 측점이 여럿이라 측점별로 뽑는다 (엔티티 id 씨앗 = "{장id}:{측점}").
        quantity_table = None
        quantity_tables: dict[str, Any] = {}
        if sheet is not None:
            for chainage in sheet["chainages"]:
                table = extract_quantity_table(f"{drawing_id}:{chainage}", request.drawing)
                if table:
                    quantity_tables[str(chainage)] = table
        elif item.kind == "cross":
            quantity_table = (
                extract_quantity_table(drawing_id, request.drawing) or request.quantity_table
            )
        # 단계 완료 기준은 횡단도(cross)만 본다. 종단도(longitudinal)는 확정 여부와 무관.
        all_confirmed = await asyncio.to_thread(
            _store_confirmed_drawing,
            project_root,
            item,
            request.drawing,
            {candidate.id for candidate in items if candidate.kind == "cross"},
            quantity_table,
            quantity_tables or None,
        )

        # 횡단도면이면 확정 단면적을 재계산한다 (재계산 실패는 도면 확정을 막지 않음).
        # 장은 담긴 측점 전부를 함께 확정한다.
        recomputed: list[tuple[int, dict[str, Any]]] = []
        cross_match = _CROSS_ID.fullmatch(drawing_id)
        pool = get_db_pool()
        targets: list[int] = []
        if sheet is not None:
            targets = list(sheet["chainages"])
        elif item.kind == "cross" and cross_match:
            targets = [int(cross_match.group(1))]
        for chainage_int in targets:
            designation = designs.get(chainage_int)
            if not designation:
                continue
            try:
                recomputed.append(
                    (
                        chainage_int,
                        await asyncio.to_thread(
                            _recompute_confirmed_design,
                            longitudinal_path,
                            f"cross_{chainage_int:05d}m",
                            designation,
                        ),
                    )
                )
            except (ValueError, KeyError, FileNotFoundError, OSError):
                logger.warning(
                    "B07 확정 단면적 재계산 실패 (도면 확정은 유지): drawing_id=%s 측점=%s",
                    drawing_id,
                    chainage_int,
                    exc_info=True,
                )

        async with pool.acquire() as connection:
            await connection.begin()
            try:
                for chainage_int, confirmed_design in recomputed:
                    await merge_cross_section_design_by_round(
                        connection,
                        route_id=route_id,
                        chainage_int=chainage_int,
                        patch=confirmed_design,
                    )
                async with connection.cursor() as cursor:
                    if all_confirmed:
                        await complete_stage(cursor, str(project_id), 4)
                    else:
                        await start_stage(cursor, str(project_id), 4)
                await connection.commit()
            except Exception:
                await connection.rollback()
                raise
        return DesignDrawingConfirmResponse(
            project_id=str(project_id),
            id=drawing_id,
            confirmed=True,
            all_confirmed=all_confirmed,
            design=recomputed[0][1] if len(recomputed) == 1 else None,
        )
    except ValueError as exc:
        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except PermissionError as exc:
        return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception("B07 도면 확정 실패: project_id=%s drawing_id=%s", project_id, drawing_id)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "상세 설계 도면을 확정하지 못했습니다."},
        )


@router.post(
    "/{project_id}/design-drawings/{drawing_id}/invalidate",
    response_model=DesignDrawingInvalidateResponse,
)
async def invalidate_design_drawing(
    project_id: UUID, drawing_id: str
) -> DesignDrawingInvalidateResponse | JSONResponse:
    """확정 도면 편집 시 B07 및 이후 단계를 미확정 상태로 되돌린다."""
    try:
        route_id, project_root, longitudinal_path, bypass = await _confirmed_source(project_id)
        # 목록은 **설계값을 넣어** 만든다 — 장 나눔이 측점 표시 폭에 따라 달라지므로,
        # 설계 없이 만들면 방금 확정한 장 id 가 목록에 없어 [수정]이 404 로 막힌다
        # (2026-09-03 실측: `cross_s00220m` 확정 후 확정 해제 불가).
        designs = await _designs_by_chainage(route_id)
        items = await asyncio.to_thread(_drawing_list, project_root, longitudinal_path, designs)
        if drawing_id not in {item.id for item in items}:
            raise FileNotFoundError("변경된 도면을 찾을 수 없습니다.")
        await asyncio.to_thread(_invalidate_drawing, project_root, drawing_id)

        cross_match = _CROSS_ID.fullmatch(drawing_id)
        stale: list[int] = []
        if cross_match:
            stale = [int(cross_match.group(1))]
        elif CROSS_SHEET_ID.fullmatch(drawing_id):
            plan = await asyncio.to_thread(
                _cross_sheet_plan, project_root, longitudinal_path, designs
            )
            sheet = next((entry for entry in plan if entry["id"] == drawing_id), None)
            stale = list(sheet["chainages"]) if sheet else []
        pool = get_db_pool()
        async with pool.acquire() as connection:
            await connection.begin()
            try:
                # 확정 도면을 편집하면 담긴 측점 설계도 잠정 상태로 되돌린다.
                for chainage_int in stale:
                    await merge_cross_section_design_by_round(
                        connection,
                        route_id=route_id,
                        chainage_int=chainage_int,
                        patch={"status": "provisional"},
                    )
                async with connection.cursor() as cursor:
                    await start_stage(cursor, str(project_id), 4)
                await connection.commit()
            except Exception:
                await connection.rollback()
                raise
        return DesignDrawingInvalidateResponse(
            project_id=str(project_id),
            id=drawing_id,
        )
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except PermissionError as exc:
        return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception(
            "B07 도면 확정 해제 실패: project_id=%s drawing_id=%s", project_id, drawing_id
        )
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "상세 설계 도면 상태를 되돌리지 못했습니다."},
        )