"""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)})