feat(485-1 B07 축척): 도면 X · Y 축척 서버 — 프로젝트마다 drawing_scales.json · GET/PUT /api/projects/{id}/drawing-scales · 엔진 drawing_scale(kind) 문맥 변수 · 종단 · 횡단(X ≠ Y) · 토적 · 전개도 · 바뀐 종류 저장본 풂 · 덧그림 mm_per_m_x · _y · 범위 밖 400 · 시험 test_280_5

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B07_DesignDetail/B07_DesignDetail_Router_Support_Scale.py

@@ -0,0 +1,191 @@
+"""B07 도면 축척(485 · 요구 280-5) — 프로젝트마다 `drawing_scales.json` 읽기 · 쓰기 · GET/PUT.
+
+엔진은 `Engine_Template.drawing_scale(kind)` 로 읽는다(문맥 변수 · 엔진 서명 그대로).
+값이 바뀐 종류의 저장본(확정)은 푼다 — 축척 다른 저장본은 쓸모없음 · 종단 가로가 바뀌면
+장 나눔 · 장 id 가 바뀌므로 그 종류 저장본을 모두 푼다. 화면은 `saved` 장 수로 먼저 묻는다.
+"""
+
+import asyncio
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from B05_Profile.B05_Profile_Engine_Sections import parse_cross_key
+from B06_Section.B06_Section_Repository import merge_cross_section_design_by_round
+from B07_DesignDetail.B07_DesignDetail_Engine_Template import (
+    DRAWING_SCALE_DEFAULTS,
+    drawing_scales_path,
+    read_drawing_scales,
+)
+from B07_DesignDetail.B07_DesignDetail_Router import _CROSS_ID, _confirmed_source
+from B07_DesignDetail.B07_DesignDetail_Router_Support import _cross_sheet_plan
+from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import (
+    _read_json,
+    _read_manifest,
+    _write_manifest,
+)
+from common_util.common_util_json import atomic_write_json
+from common_util.common_util_workflow_state import start_stage
+from config.config_db import get_db_pool
+
+router = APIRouter(prefix="/api/projects", tags=["B07 Design Detail"])
+
+# 분모 범위(정수) — 기본 · 자동 후보(토적 500~6,000 · 전개도 50~5,000)를 모두 담음.
+SCALE_RANGES: dict[str, dict[str, tuple[int, int]]] = {
+    "longitudinal": {"x": (100, 10000), "y": (10, 2000)},
+    "cross": {"x": (10, 1000), "y": (10, 1000)},
+    "mass_haul": {"x": (100, 20000), "y": (1000, 1000000)},
+    "development": {"x": (10, 10000), "y": (10, 10000)},
+}
+_KIND_LABEL = {
+    "longitudinal": "종단도",
+    "cross": "횡단도",
+    "mass_haul": "토적도",
+    "development": "전개도",
+}
+
+
+def checked_scales(scales: Any) -> dict[str, dict[str, int | None]]:
+    """{종류: {x, y}} — 분모 정수 · 범위 안 · null = 기본(자동 칸은 자동)."""
+    if not isinstance(scales, dict):
+        raise ValueError("축척은 {종류: {x, y}} 꼴로 보내 주세요.")
+    checked: dict[str, dict[str, int | None]] = {}
+    for kind, axes in scales.items():
+        if kind not in SCALE_RANGES:
+            raise ValueError(f"축척을 바꿀 수 없는 도면입니다: {kind}")
+        if not isinstance(axes, dict):
+            raise ValueError(f"{_KIND_LABEL[kind]} 축척은 {{x, y}} 꼴로 보내 주세요.")
+        row: dict[str, int | None] = {}
+        for axis, (low, high) in SCALE_RANGES[kind].items():
+            value = axes.get(axis)
+            if value is not None and (
+                isinstance(value, bool) or not isinstance(value, int) or not low <= value <= high
+            ):
+                name = "X(가로)" if axis == "x" else "Y(세로)"
+                raise ValueError(
+                    f"{_KIND_LABEL[kind]} {name} 축척은 1/{low:,}~1/{high:,} 사이 정수로 적으세요."
+                )
+            row[axis] = value
+        checked[kind] = row
+    return checked
+
+
+def saved_counts(project_root: Path) -> dict[str, int]:
+    """종류마다 저장본(확정) 장 수."""
+    counts = dict.fromkeys(SCALE_RANGES, 0)
+    for entry in _read_manifest(project_root)["drawings"].values():
+        if entry.get("confirmed") and entry.get("kind") in counts:
+            counts[entry["kind"]] += 1
+    return counts
+
+
+def scales_payload(project_root: Path) -> dict[str, Any]:
+    return {
+        "scales": read_drawing_scales(project_root),
+        "defaults": DRAWING_SCALE_DEFAULTS,
+        "ranges": {
+            kind: {axis: list(bounds) for axis, bounds in axes.items()}
+            for kind, axes in SCALE_RANGES.items()
+        },
+        "saved": saved_counts(project_root),
+    }
+
+
+def write_drawing_scales(
+    project_root: Path, longitudinal_path: Path, scales: Any
+) -> tuple[dict[str, int], list[int | float]]:
+    """보낸 종류만 바꿔 쓰고 값이 바뀐 종류의 저장본을 푼다 — (풀린 장 수, 풀린 횡단 측점)."""
+    checked = checked_scales(scales)
+    before = read_drawing_scales(project_root)
+    path = drawing_scales_path(project_root)
+    try:
+        stored = _read_json(path) if path.is_file() else {}
+    except ValueError:
+        stored = {}  # 깨진 파일 = 기본 — 새 값으로 덮음
+    atomic_write_json(path, {**stored, **checked})
+    after = read_drawing_scales(project_root)
+    changed = {kind for kind in checked if after[kind] != before[kind]}
+
+    invalidated: dict[str, int] = {}
+    stale_ids: list[str] = []
+    manifest = _read_manifest(project_root)
+    for drawing_id, entry in manifest["drawings"].items():
+        kind = entry.get("kind")
+        if entry.get("confirmed") and kind in changed:
+            entry["confirmed"] = False
+            invalidated[kind] = invalidated.get(kind, 0) + 1
+            if kind == "cross":
+                stale_ids.append(drawing_id)
+    if invalidated:
+        _write_manifest(project_root, manifest)
+
+    # 풀린 횡단 장 · 측점 도면에 담긴 측점 — 설계 상태를 잠정으로 되돌릴 몫(배치 바꿈과 같은 꼴)
+    stale: list[int | float] = []
+    sheets: dict[str, list[int | float]] | None = None
+    for drawing_id in stale_ids:
+        if match := _CROSS_ID.fullmatch(drawing_id):
+            stale.append(parse_cross_key(match.group(1)))
+            continue
+        if sheets is None:  # 장 계획은 장이 풀릴 때만 셈
+            plan = _cross_sheet_plan(project_root, longitudinal_path)
+            sheets = {sheet["id"]: sheet["chainages"] for sheet in plan}
+        stale.extend(sheets.get(drawing_id, []))
+    return invalidated, stale
+
+
+class DrawingScalesRequest(BaseModel):
+    """바꿀 종류만 — {종류: {x, y}} · 분모 정수 · null = 기본(자동)."""
+
+    scales: dict[str, Any] = Field(default_factory=dict)
+
+
+@router.get("/{project_id}/drawing-scales")
+async def get_drawing_scales(project_id: UUID) -> Any:
+    """도면 축척 — 파일 없으면 기본 · 종류마다 저장본 장 수(`saved`)."""
+    try:
+        _route_id, project_root, _longitudinal_path, _bypass = await _confirmed_source(project_id)
+        return await asyncio.to_thread(scales_payload, project_root)
+    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)})
+
+
+@router.put("/{project_id}/drawing-scales")
+async def put_drawing_scales(project_id: UUID, request: DrawingScalesRequest) -> Any:
+    """축척을 쓴다 — 값이 바뀐 종류의 저장본은 풀고 횡단이면 담긴 측점 설계를 잠정으로."""
+    try:
+        route_id, project_root, longitudinal_path, _bypass = await _confirmed_source(project_id)
+        invalidated, stale = await asyncio.to_thread(
+            write_drawing_scales, project_root, longitudinal_path, request.scales
+        )
+        if invalidated:
+            pool = get_db_pool()
+            async with pool.acquire() as connection:
+                await connection.begin()
+                try:
+                    for chainage in stale:
+                        await merge_cross_section_design_by_round(
+                            connection,
+                            route_id=route_id,
+                            chainage=chainage,
+                            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
+        payload = await asyncio.to_thread(scales_payload, project_root)
+        return {**payload, "invalidated": invalidated}
+    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)})