Merge remote-tracking branch 'origin/dev' into sub_laptop_3

# Conflicts:
#	B05_Profile/B05_Profile_UI_Structures_List.ts
#	B05_Profile/B05_Profile_UI_Structures_Panel_Types.ts
This commit is contained in:
2026-09-29 12:29:02 +09:00
33 changed files with 899 additions and 165 deletions
+33 -2
View File
@@ -98,9 +98,39 @@ def _load_design_curves(project_root: Path) -> list[dict[str, Any]]:
)
# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ────────────
#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80`
CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)"
def cross_station(chainage_m: float) -> int | float:
"""측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리."""
value = round(float(chainage_m), 2)
whole = round(value)
return int(whole) if abs(value - whole) < 0.005 else value
def cross_key(chainage_m: float) -> str:
"""이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점)."""
station = cross_station(chainage_m)
return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}"
def parse_cross_key(text: str) -> int | float:
"""`cross_key` 되읽기."""
return cross_station(float(text))
def cross_filename(chainage_m: float) -> str:
"""측점 chainage에 대응하는 횡단면 파일명(단일 규칙)."""
return f"cross_{int(round(float(chainage_m))):05d}m.json"
return f"cross_{cross_key(chainage_m)}m.json"
def find_cross_file(cross_dir: Path, chainage_m: float) -> Path:
"""읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트)."""
path = cross_dir / cross_filename(chainage_m)
legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json"
return path if path.exists() or not legacy.exists() else legacy
def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
@@ -109,8 +139,9 @@ def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고
빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다).
"""
# 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가려야 안 지워짐
valid = {
cross_filename(station["chainage_m"])
find_cross_file(cross_dir, station["chainage_m"]).name
for station in stations
if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float))
}
@@ -341,34 +341,19 @@ def generate_sections(
if 0.0 <= float(chainage) <= total
}
if extras:
# 횡단 파일명이 정수 미터 반올림이라(cross_filename) 한 정수 미터에 측점은 하나만 둔다.
# 관 · 구조물 자체 측점은 소수 그대로 — 격자 측점과 같은 정수 미터면 격자 대신 그 측점을
# 쓴다(시 · 종점 격자만 그대로 두고 라벨을 얹음). 비정규끼리 겹치면 앞선 것만 남긴다.
# ponytail: 파일명이 정수 미터인 동안의 가드. 소수 이름 규칙으로 바꾸면 지운다.
# 관 · 구조물 자리마다 그 소수 측점의 횡단을 따로 둔다(2026-09-29 사용자 확정) — 이름은
# 소수 2자리(cross_00119.80m · cross_filename). 같은 이름이 되는 0.005m 안 겹침만 하나로:
# 격자와 겹치면 격자 측점에 라벨을 얹고, 비정규끼리 겹치면 앞선 것만 남긴다.
grid = [round(float(value), 6) for value in station_chainage]
ends = {grid[0], grid[-1]}
buckets = {int(round(value)): value for value in grid}
taken: set[int] = set()
dropped: set[float] = set()
snapped: dict[float, str] = {}
for chainage in sorted(extras):
bucket = int(round(chainage))
if bucket in taken:
near = next((value for value in grid if abs(value - chainage) < 0.005), chainage)
if any(abs(near - taken) < 0.005 for taken in snapped):
continue
taken.add(bucket)
regular = buckets.get(bucket)
if regular in ends:
snapped[regular] = extras[chainage]
continue
if regular is not None:
dropped.add(regular)
snapped[chainage] = extras[chainage]
snapped[near] = extras[chainage]
extras = snapped
station_chainage = np.unique(
np.r_[
np.array([value for value in grid if value not in dropped], dtype=np.float64),
np.array(sorted(extras), dtype=np.float64),
]
np.r_[station_chainage, np.array(sorted(extras), dtype=np.float64)]
)
station_xy = _interpolate_xy(points, route_chainage, station_chainage)
tangents = np.vstack(
@@ -10,6 +10,7 @@
* chainage_m = 측점번호 × 측점간격(m) + 잔여거리(m). 측점간격은 `getInterval()`로 실시간 조회한다.
* ========================================================================== */
import { stationPartsLabel } from "../common_util/common_util_station";
import {
ESCAPE_ROUTE_DEFAULT_WIDTH_M,
ESCAPE_ROUTE_WIDTHS_M,
@@ -130,9 +131,9 @@ export function irregularStationId(id: string): string {
return `irregular:${id}`;
}
/** 측점번호+잔여거리 표기 (예: 3+18.0). */
/** 측점번호+잔여거리 표기 (예: 3+18 · 공용 표기 한 벌 — 거리는 정수로 보이기만). */
export function irregularLabel(station: IrregularStation): string {
return `${station.station}+${station.remainder.toFixed(1)}`;
return stationPartsLabel(station.station, station.remainder);
}
function field(labelText: string, input: HTMLElement): HTMLLabelElement {
@@ -24,7 +24,7 @@ import {
designHalfWidths,
summarizeCross,
} from "./B05_Profile_UI_RouteEdit_Cross_Draw";
import { formatStation } from "./B05_Profile_Util_Station";
import { stationLabel as formatStation } from "../common_util/common_util_station";
export interface CrossPreviewParams {
projectId: string;
@@ -13,7 +13,7 @@
import { fetchRouteElevations } from "./B05_Profile_Api_Replan";
import { routePointAtScreen, type RoutePointHit } from "./B05_Profile_UI_RouteEdit_Input";
import { formatStation } from "./B05_Profile_Util_Station";
import { stationLabel as formatStation } from "../common_util/common_util_station";
type Vertex = [number, number];
@@ -26,7 +26,7 @@ import { contourBandRect } from "./B05_Profile_UI_RouteEdit_Input";
import type { CrossWatch } from "./B05_Profile_UI_RouteEdit_Cross";
import { stationFrameAt } from "./B05_Profile_UI_RouteEdit_Frame";
import type { MeasureMark } from "./B05_Profile_UI_RouteEdit_Measure";
import { formatStation } from "./B05_Profile_Util_Station";
import { stationLabel as formatStation } from "../common_util/common_util_station";
/** 노드 반지름(px). */
const NODE_R = 4;
@@ -14,7 +14,7 @@ import {
type StructureType,
} from "./B05_Profile_Api_Structures";
import type { PipeFacilityItem } from "./B05_Profile_UI_Structures_Panel";
import { formatStation } from "./B05_Profile_Util_Station";
import { stationLabel as formatStation } from "../common_util/common_util_station";
import { readSummaryItem, summaryTooltip } from "../common_util/common_util_summary_item";
export interface StructureListParams {
@@ -20,7 +20,7 @@ import {
type StructureType,
} from "./B05_Profile_Api_Structures";
import { numberDuplicateNames } from "./B05_Profile_UI_Structures_List";
import { formatStation } from "./B05_Profile_Util_Station";
import { stationLabel as formatStation } from "../common_util/common_util_station";
/** 마크를 잡았다고 볼 여유(px) — 시작점에서 이만큼 **벗어나기 전에는** 이동으로
* 보지 않고 시각적으로도 움직이지 않는다. 예전 10px 누적 방식은 손떨림이 쌓여
+10 -27
View File
@@ -38,13 +38,19 @@ import {
loadPipeForm as loadPipeFormInto,
} from "./B05_Profile_UI_Structures_Panel_Commit";
import { renderStructureList } from "./B05_Profile_UI_Structures_List";
import { formatStation } from "./B05_Profile_Util_Station";
import { stationLabel as formatStation } from "../common_util/common_util_station";
/** 공개 타입·표시 상수는 700줄 제한으로 `_Structures_Panel_Types` 로 옮겼다 —
* 옛 임포트 경로가 그대로 동작하도록 여기서 다시 내보낸다(2026-09-04). */
export { GROUP_LABELS } from "./B05_Profile_UI_Structures_Panel_Types";
export type { PipeFacilityItem, StructuresSection } from "./B05_Profile_UI_Structures_Panel_Types";
import { DEFAULT_INTERVAL_LENGTH_M, GROUP_LABELS } from "./B05_Profile_UI_Structures_Panel_Types";
import {
DEFAULT_INTERVAL_LENGTH_M,
GROUP_LABELS,
hasComputedRange,
readBeforeOption,
readLengthOption,
} from "./B05_Profile_UI_Structures_Panel_Types";
import type {
PipeFacilityItem,
StructuresCallbacks,
@@ -92,35 +98,12 @@ export function createStructuresSection(
return type.managed_by ? (type.type_id as PipeFacility) : null;
}
/** 구간을 직접 입력하지 않는 타입인가 — C군 사면안정 5종처럼 B05 길이 옵션이
* 있으면 시작·종료 입력 칸을 없애고 기준 측점 + 길이로 범위를 계산해 보여만
* 준다(2026-08-19 사용자 지시). */
function hasComputedRange(type: StructureType | null): boolean {
return (
!!type &&
!type.managed_by &&
type.placement === "interval" &&
type.options.some((option) => option.key === "length_m" && isB05Option(option))
);
}
/** 길이 옵션(length_m) 값(m). 비었거나 0 이하면 null — 범위를 계산할 수 없다. */
function readLengthM(): number | null {
const entry = optionInputs.find((input) => input.key === "length_m");
if (!entry || entry.isEmpty()) return null;
const length = Number(entry.read());
return Number.isFinite(length) && length > 0 ? length : null;
return readLengthOption(optionInputs);
}
/** 기준측점 전(before_m, m) — 기준 측점에서 시점 쪽으로 물러나는 몫. 비우면 0
* (= 기준이 곧 시작). 총길이를 넘길 수 없다 — 넘치면 총길이로 자른다(후 0). */
function readBeforeM(): number {
const entry = optionInputs.find((input) => input.key === "before_m");
if (!entry || entry.isEmpty()) return 0;
const before = Number(entry.read());
if (!Number.isFinite(before) || before < 0) return 0;
const length = readLengthM();
return length !== null && before > length ? length : before;
return readBeforeOption(optionInputs);
}
/** 기준측점 전·후 중 나중에 고친 쪽 — 길이가 바뀔 때 이쪽 값을 지키며 재배분한다. */
@@ -13,7 +13,11 @@ import type {
PipeSource,
} from "../B04_PreProcess/B04_PreProcess_Api_Fetch";
import type { FacilityAttributes, FacilityOptionsForm } from "./B05_Profile_UI_Drainage_Facility";
import type { StructureInstance, StructureType } from "./B05_Profile_Api_Structures";
import {
isB05Option,
type StructureInstance,
type StructureType,
} from "./B05_Profile_Api_Structures";
import type { SummaryItem } from "../common_util/common_util_summary_item";
/** 구조물군 표시 이름. 리스트 기호(A~G)만으로는 무엇인지 알기 어렵다.
@@ -124,3 +128,36 @@ export interface StructuresCallbacks {
*/
onReveal?: () => void;
}
type OptionEntry = { key: string; read: () => string | number; isEmpty: () => boolean };
/** 길이 옵션(length_m) 값(m). 비었거나 0 이하면 null — 범위를 계산할 수 없다. */
export function readLengthOption(inputs: OptionEntry[]): number | null {
const entry = inputs.find((input) => input.key === "length_m");
if (!entry || entry.isEmpty()) return null;
const length = Number(entry.read());
return Number.isFinite(length) && length > 0 ? length : null;
}
/** 기준측점 전(before_m, m) — 기준 측점에서 시점 쪽으로 물러나는 몫. 비우면 0
* (= 기준이 곧 시작). 총길이를 넘길 수 없다 — 넘치면 총길이로 자른다(후 0). */
export function readBeforeOption(inputs: OptionEntry[]): number {
const entry = inputs.find((input) => input.key === "before_m");
if (!entry || entry.isEmpty()) return 0;
const before = Number(entry.read());
if (!Number.isFinite(before) || before < 0) return 0;
const length = readLengthOption(inputs);
return length !== null && before > length ? length : before;
}
/** 구간을 직접 입력하지 않는 타입인가 — C군 사면안정 5종처럼 B05 길이 옵션이
* 있으면 시작·종료 입력 칸을 없애고 기준 측점 + 길이로 범위를 계산해 보여만
* 준다(2026-08-19 사용자 지시). */
export function hasComputedRange(type: StructureType | null): boolean {
return (
!!type &&
!type.managed_by &&
type.placement === "interval" &&
type.options.some((option) => option.key === "length_m" && isB05Option(option))
);
}
+2 -7
View File
@@ -4,7 +4,8 @@
*
* 구조물 위치는 저장은 누가거리(m)로 하되 **화면 입출력은 측점번호+잔여거리**로 한다
* (2026-08-17 사용자 확정). 근거: 중심선측량 측점 20m + 구조물 지점 보조말뚝
* (지식DB 설계제원_총괄 §8). 표기 예: "3+18.0" = 측점 3번에서 18.0m.
* (지식DB 설계제원_총괄 §8). 표기 예: "3+18" = 측점 3번에서 18m — 화면 표기는
* `common_util_station.stationLabel` 한 벌(거리는 정수로 보이기만 · 값은 소수).
*
* 변환식은 비정규 측점 UI(`B05_Profile_UI_IrregularStations.ts`)와 같다 —
* chainage_m = 측점번호 × 측점간격(m) + 잔여거리(m). 측점간격은 호출 시점에 받는다
@@ -33,12 +34,6 @@ export function chainageToStation(
return { station, remainder };
}
/** 누가거리 → "3+18.0" 표기. */
export function formatStation(chainageM: number, intervalM: number): string {
const { station, remainder } = chainageToStation(chainageM, intervalM);
return `${station}+${remainder.toFixed(1)}`;
}
/** "3+18.0" · "3 + 18" · "76.5"(누가거리 직접) 입력을 누가거리로 해석한다.
* 해석 불가하면 null — 호출부가 칸을 붉히고 진행을 멈춘다. */
export function parseStationText(text: string, intervalM: number): number | null {
@@ -19,7 +19,7 @@ from typing import Any, Callable
import numpy as np
from B05_Profile.B05_Profile_Engine_Sections import cross_filename
from B05_Profile.B05_Profile_Engine_Sections import find_cross_file
from B06_Section.B06_Section_Engine_Design import (
_SLOPE_CLOSE_TOLERANCE_M,
compute_cross_design,
@@ -175,7 +175,7 @@ def extend_unclosed_sections(
recomputed[key] = design[key]
section["samples"] = samples
section["design"] = recomputed
cross_path = cross_dir / cross_filename(chainage)
cross_path = find_cross_file(cross_dir, chainage)
if cross_path.is_file():
stored = json.loads(cross_path.read_text(encoding="utf-8"))
atomic_write_json(cross_path, {**stored, "samples": samples})
+10 -10
View File
@@ -406,23 +406,23 @@ async def merge_cross_section_design_by_round(
connection: aiomysql.Connection,
*,
route_id: int,
chainage_int: int,
chainage: float,
patch: dict[str, Any],
) -> bool:
"""정수 chainage(m) 측점의 data.design에 patch를 병합 저장한다.
"""측점(도면 id 측점 — 정수 m 또는 소수 2자리)의 data.design에 patch를 병합 저장한다.
B07 도면 확정(전체 설계 재계산 결과) 및 확정 해제(status 되돌림)에서 사용한다.
도면 ID가 정수 m라 ROUND로 매칭한다. 갱신 여부를 반환한다.
이름 규칙(cross_key)의 0.005m 안으로 매칭한다. 갱신 여부를 반환한다.
"""
async with connection.cursor() as cursor:
await cursor.execute(
"""
SELECT id, data FROM cross_sections
WHERE route_id = %s AND ROUND(chainage_m) = %s
WHERE route_id = %s AND ABS(chainage_m - %s) < 0.005
ORDER BY id DESC
LIMIT 1
""",
(route_id, chainage_int),
(route_id, chainage),
)
row = await cursor.fetchone()
if not row:
@@ -445,21 +445,21 @@ async def merge_cross_section_design_by_round(
async def get_cross_section_design(
connection: aiomysql.Connection, route_id: int, chainage_int: int
connection: aiomysql.Connection, route_id: int, chainage: float
) -> dict[str, Any] | None:
"""정수 chainage(m)에 해당하는 측점의 잠정 설계(data.design)를 반환한다.
"""도면 id 측점(cross_{정수m}m · cross_{소수2자리}m)의 잠정 설계(data.design)를 반환한다.
B08이 도면 ID(cross_{정수m}m)로 조회하므로 ROUND로 근사 매칭한다.
이름 규칙(cross_key)의 0.005m 안으로 매칭한다.
"""
async with connection.cursor() as cursor:
await cursor.execute(
"""
SELECT data FROM cross_sections
WHERE route_id = %s AND ROUND(chainage_m) = %s
WHERE route_id = %s AND ABS(chainage_m - %s) < 0.005
ORDER BY id DESC
LIMIT 1
""",
(route_id, chainage_int),
(route_id, chainage),
)
row = await cursor.fetchone()
if not row or not row[0]:
+3 -3
View File
@@ -5,7 +5,7 @@ import logging
from pathlib import Path
from typing import Any
from B05_Profile.B05_Profile_Engine_Sections import cross_filename
from B05_Profile.B05_Profile_Engine_Sections import find_cross_file
from B05_Profile.B05_Profile_Structures_Repository import load_structures
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
from B06_Section.B06_Section_Engine_Culvert import load_culvert_sets
@@ -309,7 +309,7 @@ def read_cross_design_inputs(
design_elevation = design_elevation_from_longitudinal(longitudinal, chainage_m)
suggested = pavement.get(round(chainage_m, 3), False)
cross_dir = longitudinal_path.parent.parent / "cross_sections"
cross_path = (cross_dir / cross_filename(chainage_m)).resolve()
cross_path = find_cross_file(cross_dir, chainage_m).resolve()
if cross_dir.resolve() not in cross_path.parents or not cross_path.is_file():
raise FileNotFoundError("해당 측점의 횡단 상세 파일을 찾을 수 없습니다.")
cross = json.loads(cross_path.read_text(encoding="utf-8"))
@@ -492,7 +492,7 @@ def compute_default_designs(
ford_drops = ford_surface_drops(root)
results: list[tuple[float, dict[str, Any]]] = []
for chainage_m in chainages:
cross_path = cross_dir / cross_filename(chainage_m)
cross_path = find_cross_file(cross_dir, chainage_m)
if not cross_path.is_file():
continue
try:
@@ -9,6 +9,7 @@
import re
from typing import Any
from B05_Profile.B05_Profile_Engine_Sections import cross_key
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
CROSS_MM,
CROSS_TABLE_LAYER_ID,
@@ -38,7 +39,7 @@ from config.config_system import DRAWING_SCALE_CROSS
# 장 도면 id — 측점 도면(cross_00020m)과 겹치지 않게 `s`를 끼운다(cross_s00020m).
# 옛 순번 형식(cross_s01)도 읽어 준다 — 이미 확정한 매니페스트가 그 이름을 갖고 있다.
CROSS_SHEET_ID = re.compile(r"cross_s(\d{2,})m?")
CROSS_SHEET_ID = re.compile(r"cross_s(\d{2,}(?:\.\d{2})?)m?")
# 블록 사이 여백(mm)과 테두리 여유 — build_cross_drawing의 테두리 값과 맞춘다.
# 여백 0 = 이웃 칸의 테두리가 맞닿는다(2026-09-04 사용자 지시 — 박스 사이 간격 제거).
@@ -151,7 +152,7 @@ def _slots(cell_w: float, cell_h: float, rows: int, count: int) -> list[list[flo
return slots
def plan_cross_sheets(blocks: list[tuple[int, float, float]]) -> list[dict[str, Any]]:
def plan_cross_sheets(blocks: list[tuple[int | float, float, float]]) -> list[dict[str, Any]]:
"""(측점, 폭, 높이) 목록을 A1 장으로 나눈다.
한 장 안의 칸은 모두 같은 크기(그 장 최대 블록 기준)이고 빈 곳은 여백으로 둔다.
@@ -170,7 +171,7 @@ def plan_cross_sheets(blocks: list[tuple[int, float, float]]) -> list[dict[str,
# id는 **시작 측점**으로 짓는다. 순번(`cross_s01`)으로 지으면 B06 설계가
# 바뀌어 한 장에 담기는 측점 수가 달라졌을 때 같은 이름이 다른 구간을
# 가리키고, 옛 확정 표시가 그대로 새 구간에 붙는다(2026-09-01 지적).
"id": f"cross_s{chainages[0]:05d}m",
"id": f"cross_s{cross_key(chainages[0])}m",
"number": len(sheets) + 1,
"chainages": chainages,
"rows": max(rows, 1),
+20 -15
View File
@@ -13,6 +13,11 @@ from fastapi import APIRouter
from fastapi.responses import JSONResponse
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Engine_Sections import (
CROSS_KEY_PATTERN,
cross_station,
parse_cross_key,
)
from B06_Section.B06_Section_Repository import (
get_confirmed_route_context,
get_cross_section_design,
@@ -65,7 +70,7 @@ 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$")
_CROSS_ID = re.compile(rf"^cross_{CROSS_KEY_PATTERN}m$")
_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")
@@ -249,13 +254,13 @@ async def title_block_fields(project_id: UUID) -> dict[str, str]:
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]]:
async def _designs_by_chainage(route_id: int) -> dict[int | float, 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"]
cross_station(row["chainage_m"]): row["design"]
for row in rows
if isinstance(row, dict) and isinstance(row.get("design"), dict)
}
@@ -304,7 +309,7 @@ async def get_design_drawing(
pool = get_db_pool()
async with pool.acquire() as connection:
design = await get_cross_section_design(
connection, route_id, int(cross_match.group(1))
connection, route_id, parse_cross_key(cross_match.group(1))
)
source_design = design
elif CROSS_SHEET_ID.fullmatch(drawing_id):
@@ -448,25 +453,25 @@ async def confirm_design_drawing(
# 트림과 사용자 입력(측구 끔)이 사라졌다(2026-09-14 936be972 실측 62측점 · 브레인 ②).
cross_match = _CROSS_ID.fullmatch(drawing_id)
pool = get_db_pool()
targets: list[int] = []
targets: list[int | float] = []
if sheet is not None:
targets = list(sheet["chainages"])
elif item.kind == "cross" and cross_match:
targets = [int(cross_match.group(1))]
targets = [parse_cross_key(cross_match.group(1))]
confirmed_designs = [
(chainage_int, {**designs[chainage_int], "status": "confirmed"})
for chainage_int in targets
if designs.get(chainage_int)
(chainage, {**designs[chainage], "status": "confirmed"})
for chainage in targets
if designs.get(chainage)
]
async with pool.acquire() as connection:
await connection.begin()
try:
for chainage_int, _design in confirmed_designs:
for chainage, _design in confirmed_designs:
await merge_cross_section_design_by_round(
connection,
route_id=route_id,
chainage_int=chainage_int,
chainage=chainage,
patch={"status": "confirmed"},
)
async with connection.cursor() as cursor:
@@ -519,9 +524,9 @@ async def invalidate_design_drawing(
await asyncio.to_thread(_invalidate_drawing, project_root, drawing_id)
cross_match = _CROSS_ID.fullmatch(drawing_id)
stale: list[int] = []
stale: list[int | float] = []
if cross_match:
stale = [int(cross_match.group(1))]
stale = [parse_cross_key(cross_match.group(1))]
elif CROSS_SHEET_ID.fullmatch(drawing_id):
plan = await asyncio.to_thread(
_cross_sheet_plan, project_root, longitudinal_path, designs
@@ -533,11 +538,11 @@ async def invalidate_design_drawing(
await connection.begin()
try:
# 확정 도면을 편집하면 담긴 측점 설계도 잠정 상태로 되돌린다.
for chainage_int in stale:
for chainage in stale:
await merge_cross_section_design_by_round(
connection,
route_id=route_id,
chainage_int=chainage_int,
chainage=chainage,
patch={"status": "provisional"},
)
async with connection.cursor() as cursor:
@@ -9,6 +9,12 @@ import re
from pathlib import Path
from typing import Any
from B05_Profile.B05_Profile_Engine_Sections import (
CROSS_KEY_PATTERN,
cross_station,
find_cross_file,
parse_cross_key,
)
from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
DRAWING_FORMAT,
@@ -94,7 +100,7 @@ logger = logging.getLogger(__name__)
_STAGE_DIR = "B07_DesignDetail"
_CROSS_ID = re.compile(r"^cross_(\d+)m$")
_CROSS_ID = re.compile(rf"^cross_{CROSS_KEY_PATTERN}m$")
_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")
# 노선 전장에 한 장씩만 나오는 도면 — 라우터가 원본 자료를 따로 실어 넘긴다.
MASS_HAUL_ID = "mass_haul"
@@ -107,7 +113,7 @@ CROSS_STANDARD_ID = "cross_standard"
def _drawing_list(
project_root: Path,
longitudinal_path: Path,
designs: dict[int, dict[str, Any]] | None = None,
designs: dict[int | float, dict[str, Any]] | None = None,
) -> list[DesignDrawingItem]:
longitudinal = _read_json(longitudinal_path)
station_by_chainage = _station_map(longitudinal)
@@ -226,7 +232,7 @@ def _store_drawing_numbers(project_root: Path, drawings: list[DesignDrawingItem]
def _cross_sheet_plan(
project_root: Path,
longitudinal_path: Path,
designs: dict[int, dict[str, Any]] | None = None,
designs: dict[int | float, dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""측점 도면을 A1 장으로 나눈 계획 — 목록과 도면 생성이 같은 결과를 쓴다.
@@ -239,8 +245,9 @@ def _cross_sheet_plan(
match = _CROSS_ID.fullmatch(path.stem)
if not match:
continue
chainage = int(match.group(1))
source = _read_json(path)
# 측점은 파일 속 값으로 — 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 이름으로 못 앎
chainage = cross_station(source.get("chainage_m") or parse_cross_key(match.group(1)))
design = (designs or {}).get(chainage)
design_line = _cross_design_line(longitudinal_path, source, design)
width, height = section_block_size(source, design_line)
@@ -352,11 +359,11 @@ def _cross_design_line(
def _cross_section_input(
longitudinal_path: Path,
longitudinal: dict[str, Any],
chainage: int,
chainage: int | float,
design: dict[str, Any] | None,
) -> dict[str, Any] | None:
"""한 측점의 장 배치 입력(원본·계획선·수량표·제목)을 만든다."""
path = longitudinal_path.parent.parent / "cross_sections" / f"cross_{chainage:05d}m.json"
path = find_cross_file(longitudinal_path.parent.parent / "cross_sections", chainage)
if not path.is_file():
return None
source = _read_json(path)
@@ -565,9 +572,12 @@ def _read_drawing(
if CROSS_SHEET_ID.fullmatch(drawing_id):
return _read_cross_sheet(project_root, longitudinal_path, drawing_id, stored_design)
if not _CROSS_ID.fullmatch(drawing_id):
cross_match = _CROSS_ID.fullmatch(drawing_id)
if not cross_match:
raise ValueError("올바르지 않은 도면 ID입니다.")
path = longitudinal_path.parent.parent / "cross_sections" / f"{drawing_id}.json"
path = find_cross_file(
longitudinal_path.parent.parent / "cross_sections", parse_cross_key(cross_match.group(1))
)
if not path.is_file():
raise FileNotFoundError("요청한 횡단도를 찾을 수 없습니다.")
source = _read_json(path)
@@ -9,7 +9,7 @@ import re
from pathlib import Path
from typing import Any
from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files
from B05_Profile.B05_Profile_Engine_Sections import cross_station, prune_stale_cross_files
logger = logging.getLogger(__name__)
@@ -47,12 +47,12 @@ def _cross_files(longitudinal_path: Path, longitudinal: dict[str, Any]) -> list[
return files
def _station_map(longitudinal: dict[str, Any]) -> dict[int, dict[str, Any]]:
def _station_map(longitudinal: dict[str, Any]) -> dict[int | float, dict[str, Any]]:
stations = longitudinal.get("stations", [])
if not isinstance(stations, list):
return {}
return {
round(float(station.get("chainage_m", 0))): station
cross_station(station.get("chainage_m", 0)): station
for station in stations
if isinstance(station, dict)
}
@@ -0,0 +1,130 @@
"""M02 층 집계표 항목 — 회사 · 개인 층 빈 뼈대 · 항목 목록 · 항목 복사(PLAN 21-1 · 21-2).
항목 = 구조물집계표조합(SS) 한 줄 — 그 열의 일위대가(UA) · 자재조합(UM) ·
도번 SD · 면적번 SA · 재료번 SM.
층 폴더(시스템 · 회사 · 개인)는 모양이 같아 저장소 함수를 `store.at(층 폴더)` 안에서 그대로 부름.
권한은 부르는 쪽(`M02_MasterTemplete_Router_Items.py`) 몫 — 여기는 폴더와 파일만.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_MasterTemplete_Matset as matset
from M02_MasterTemplete import M02_MasterTemplete_Store as store
# 21-1 목록 칸 — SS 줄에서 그대로(키는 `key`)
ITEM_FIELDS = (
"열",
"종류",
"분류1",
"분류2",
"분류3",
"단위",
"일위대가",
"자재조합",
"도번",
"면적번",
"재료번",
)
def ensure(folder: Path) -> bool:
"""회사 · 개인 층 첫 진입 — 빈 집계표(열 0 · 층 이름 · 줄 머리는 시스템 것) ·
조합 장부(`다음` 1) · 자재 조합 빈 표 · 종류 폴더.
이미 집계표가 있으면 아무것도 안 함(시스템 층은 늘 있음)."""
table = Path(folder) / "table" / f"{store.STRUCTURE_TABLE}.json"
if table.is_file():
return False
system = json.loads((store.FOLDER / "table" / f"{store.STRUCTURE_TABLE}.json").read_bytes())
doc = {
**system,
"열": [],
"줄": [{**r, "값": {}} for r in system.get("줄", []) if isinstance(r, dict)],
"합계줄": [],
"보기": {},
}
with store.at(folder):
for kind in store.KINDS:
(Path(folder) / kind).mkdir(parents=True, exist_ok=True)
with store._LOCK:
if table.is_file(): # 그 사이 다른 요청이 만듦
return False
atomic_write_json(table, doc)
store.sync_combos()
mat = Path(folder) / matset.FILE
if not mat.is_file():
atomic_write_json(mat, matset._load()[0])
return True
def items(folder: Path) -> list[dict[str, Any]]:
"""그 층 항목 — SS 줄마다 `{key, 열, 종류, 분류1~3, 단위, 일위대가, 자재조합,
도번, 면적번, 재료번}`."""
with store.at(folder):
rows = store.read_combos().get("줄", [])
return [{"key": r.get("키"), **{k: r.get(k) for k in ITEM_FIELDS}} for r in rows]
def _fresh_id(want: str, used: set[str]) -> str:
out, n = want, 2
while out in used:
out, n = f"{want}_{n}", n + 1
return out
def copy_item(source: Path, target: Path, key: str, origin: dict[str, Any]) -> dict[str, Any]:
"""다른 층 항목(SS `key`) → `target` 층 새 열 하나 — 머리 · 단위 · 일위대가 · 바인딩 그대로 ·
도면 · 면적산출 · 재료산출 내용(알약 연결 포함) · 자재 조합 줄(새 UM) 복사 ·
새 번호 · 새 SS 키 ·
열 `본뜬` = `origin`(원래 층 · 주인 · 키). 돌려줌 `{key, 열, 이름}`."""
with store.at(source):
row = next((r for r in store.read_combos().get("줄", []) if r.get("키") == key), None)
if row is None:
raise store.StoreError(404, f"없는 항목 「{key}」")
col_id = row["열"]
col = next(c for c in store._structure_columns() if c["id"] == col_id)
docs = {}
for kind, _, _, _ in store.PARTS:
try:
docs[kind] = store.read(kind, col_id)["문서"]
except store.StoreError:
pass # 없는 파일은 새 층에서 빈 문서로 섬
um = None
if col.get("자재조합"):
try:
um = matset.one(col["자재조합"])
except store.StoreError:
um = None # 가리키던 줄이 없으면 비움
ensure(target)
with store.at(target):
new_um = None
if um:
names = matset.listing()["줄"]
taken = {str(r.get("이름") or "") for r in names}
name, n = str(um.get("이름") or key), 2
base = name
while name in taken:
name, n = f"{base} ({n})", n + 1
got = {k: um.get(k) for k in ("담은자재", "소유", "비고")}
new_um = matset.create({**got, "이름": name, "본뜬키": None})["키"]
table = store.read("table", store.STRUCTURE_TABLE)
doc = table["문서"]
new_id = _fresh_id(col_id, {c.get("id") for c in doc.get("열", []) if isinstance(c, dict)})
made = {**json.loads(json.dumps(col)), "id": new_id, "자재조합": new_um, "본뜬": origin}
doc["열"] = [*doc.get("열", []), made]
store.write("table", store.STRUCTURE_TABLE, table["판"], doc) # 빈 세 파일 · 번호 · SS
for kind, got in docs.items():
number = store._KEY[kind]
have = store.read(kind, new_id)["문서"]
body = {k: v for k, v in got.items() if k != "계산값"}
body = {**body, "열": new_id, number: have.get(number)}
if kind in store.SHEETS:
body = store._server_values(body)
with store._LOCK:
atomic_write_json(store._path(kind, new_id), body)
ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id)
return {"key": ss["키"], "열": new_id, "이름": " ".join(h for h in store._heads(made) if h)}
@@ -22,7 +22,7 @@ _HEAD = "UM"
def _load() -> tuple[dict[str, Any], str]:
"""(표, 판) — 파일이 없으면 빈 표 · 판 빈 글."""
path = store.FOLDER / FILE
path = store.root() / FILE
if not path.is_file():
return {"그룹": mmc.GROUP, "판": "2026", "비고": [], "다음": 1, "줄": []}, ""
raw = path.read_bytes()
@@ -95,7 +95,7 @@ def _save(doc: dict, version: str) -> str:
"""틀 검사 → 판이 같을 때만 씀 → 새 판."""
if errors := mmc.check_form(FILE, doc):
raise store.StoreError(422, {"errors": errors})
path = store.FOLDER / FILE
path = store.root() / FILE
with store._LOCK:
have = store.version_of(path.read_bytes()) if path.is_file() else ""
if have != (version or ""):
@@ -0,0 +1,263 @@
"""M02 층 집계표 API — `/api/m02/layers/{layer}/…` · 항목 목록 · 항목 복사(PLAN 21-1 · 21-2).
⚠ 등록은 `main.py`(로그인만) · 층 권한은 여기서(`Router_Layers._layer_dir` 와 같은 규칙):
system 읽기 = 로그인 누구나(항목 고르기) · 쓰기 = SYSTEM_ADMIN
company 읽기 = 같은 회사 · 쓰기 = 회사 관리자(ADMIN · 마스터 · SYSTEM_ADMIN)
personal 읽기 · 쓰기 = 본인(SYSTEM_ADMIN 은 누구나) · 회사 없음 = 쓰기 409 「회사부터 등록」
옛 길(`/api/m02/structures/…` 등 · `M02_MasterTemplete_Router.py`)은 시스템 층 별칭으로 그대로.
회사 · 개인 층은 처음 부를 때 빈 뼈대를 만듦(`Items.ensure`).
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Literal
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_MasterTemplete_Branch as branch
from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Matset as matset
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_MasterTemplete_Router import (
BranchBody,
MaterialBranchBody,
MatsetBody,
UseComboBody,
)
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"])
Layer = Literal["system", "company", "personal"]
Session = dict[str, Any]
NO_COMPANY = "회사부터 등록 — 회사 없이는 회사 · 개인 구조물 집계표를 만들 수 없음"
class CopyItemBody(BaseModel):
from_layer: Literal["system", "company"]
from_owner: str | None = None
key: str
def _admin(session: Session) -> bool:
return session.get("role") == "SYSTEM_ADMIN"
def _company_admin(session: Session) -> bool:
return session.get("role") in ("ADMIN", "SYSTEM_ADMIN") or bool(session.get("is_master"))
def _owner(session: Session, layer: str, owner: str | None) -> str | None:
"""층 주인 글 — 회사 `{회사}` · 개인 `{회사}/{사용자}` · 시스템 None ·
남의 것은 SYSTEM_ADMIN 만."""
if layer == "system":
return None
company, user = session.get("company_id"), session.get("user_id")
mine = str(company) if layer == "company" else f"{company}/{user}"
if owner in (None, "", mine):
return None if company is None else mine
if not _admin(session):
raise HTTPException(status_code=403, detail="다른 회사 · 다른 사람의 구조물 집계표입니다.")
return owner
def _folder(layer: str, owner: str | None) -> Path:
if layer == "system":
return store.FOLDER
try:
if layer == "company":
return layers.company_dir(owner)
company, user = str(owner).split("/", 1)
return layers.personal_dir(company, user)
except ValueError as error:
raise HTTPException(status_code=400, detail=f"층 주인 모양 — {error}") from error
def _root(session: Session, layer: str, owner: str | None = None, write: bool = False) -> Path:
"""층 폴더(권한까지) — 회사 · 개인은 없으면 빈 뼈대를 만듦. 회사 없는 사람: 쓰기 409."""
if layer == "system":
if write and not _admin(session):
raise HTTPException(status_code=403, detail="마스터 템플릿은 시스템 관리자만 고침.")
return store.FOLDER
who = _owner(session, layer, owner)
if who is None:
raise HTTPException(status_code=409, detail=NO_COMPANY)
if write and layer == "company" and not _company_admin(session):
raise HTTPException(
status_code=403, detail="회사용은 회사 관리자만 고침 — 개인용으로 복사해서 씀."
)
folder = _folder(layer, who)
items_store.ensure(folder)
return folder
def _call(folder: Path, fn, *args):
with store.at(folder):
try:
return fn(*args)
except store.StoreError as e:
raise HTTPException(status_code=e.status, detail=e.detail) from e
# ── 항목 목록 · 복사 ───────────────────────────────────
def _listing(session: Session, layer: str, owner: str | None) -> dict[str, Any]:
who = None if layer == "system" else _owner(session, layer, owner)
if layer != "system" and who is None:
return {"layer": layer, "owner": None, "items": []} # 회사 없음 = 빈 그룹
folder = _root(session, layer, who)
return {"layer": layer, "owner": who, "items": items_store.items(folder)}
@router.get("/items")
def get_items(
layer: Layer = "system",
owner: str | None = None,
all: bool = False, # noqa: A002 — 21-1 약속 이름 `?all=1`
session: Session = Depends(verify_session),
):
"""21-1 — 한 층이면 `[{key, 열, 종류, 분류1~3, 단위, 일위대가, 자재조합, 도번, 면적번,
재료번}]` ·
`?all=1` 이면 세 층 `[{layer, owner, items}]`(회사 · 개인은 세션 것 · 회사 없으면 빈 목록)."""
if all:
return [_listing(session, one, None) for one in ("system", "company", "personal")]
return _listing(session, layer, owner)["items"]
@router.post("/layers/personal/items/copy")
def copy_item(body: CopyItemBody, session: Session = Depends(verify_session)) -> dict:
"""회사 · 마스터 항목 → 내 개인 층 새 항목(열 · 세 파일 · 자재 조합 줄 · 새 SS 키 · `본뜬`)."""
target = _root(session, "personal", None, write=True)
source = _root(session, body.from_layer, body.from_owner)
owner = (
None if body.from_layer == "system" else _owner(session, body.from_layer, body.from_owner)
)
origin = {"layer": body.from_layer, "owner": owner, "key": body.key}
try:
return items_store.copy_item(source, target, body.key, origin)
except store.StoreError as e:
raise HTTPException(status_code=e.status, detail=e.detail) from e
# ── 층 길 — 옛 시스템 길과 같은 모양 ─────────────────────
@router.get("/layers/{layer}/structures/numbers")
def get_numbers(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
return _call(_root(session, layer, owner), store.structure_numbers)
@router.get("/layers/{layer}/structure-combos")
def get_combos(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
return _call(_root(session, layer, owner), store.read_combos)
@router.post("/layers/{layer}/structures/{column}")
def post_structure(
layer: Layer, column: str, owner: str | None = None, session: Session = Depends(verify_session)
):
return _call(_root(session, layer, owner, write=True), store.create_structure, column)
@router.post("/layers/{layer}/structures/{column}/combo")
def post_combo(
layer: Layer,
column: str,
body: BranchBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
"""일위대가 조합 갈라 내기 — 새 UA 는 M01 시스템 마스터에 생기므로 SYSTEM_ADMIN 만."""
folder = _root(session, layer, owner, write=True)
if not _admin(session):
raise HTTPException(status_code=403, detail="일위대가 조합은 시스템 관리자만 새로 만듦.")
return _call(
folder, branch.branch_combo, column, body.본뜬키, body.담은로직, body.이름, body.판
)
@router.post("/layers/{layer}/structures/{column}/material-combo")
def post_material_combo_branch(
layer: Layer,
column: str,
body: MaterialBranchBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
folder = _root(session, layer, owner, write=True)
args = (column, body.본뜬키, body.담은자재, body.이름, body.판)
return _call(folder, branch.branch_material_combo, *args)
@router.post("/layers/{layer}/structures/{column}/use-combo")
def post_use_combo(
layer: Layer,
column: str,
body: UseComboBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
folder = _root(session, layer, owner, write=True)
return _call(folder, branch.use_combo, column, body.종류, body.키, body.판)
@router.post("/layers/{layer}/{kind}/{column}/copy")
def post_copy(
layer: Layer,
kind: Literal["structures", "area", "material"],
column: str,
source: str = Query(alias="from"),
owner: str | None = None,
session: Session = Depends(verify_session),
):
folder = _root(session, layer, owner, write=True)
part = "structure" if kind == "structures" else kind
return _call(folder, store.copy_from, part, column, source)
@router.get("/layers/{layer}/material-combos")
def get_matsets(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
return _call(_root(session, layer, owner), matset.listing)
@router.get("/layers/{layer}/material-combos/{key}")
def get_matset(
layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session)
):
return _call(_root(session, layer, owner), matset.one, key)
@router.post("/layers/{layer}/material-combos")
def post_matset(
layer: Layer,
body: MatsetBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
return _call(_root(session, layer, owner, write=True), matset.create, body.줄, body.판)
@router.put("/layers/{layer}/material-combos/{key}")
def put_matset(
layer: Layer,
key: str,
body: MatsetBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
return _call(_root(session, layer, owner, write=True), matset.edit, key, body.줄, body.판)
@router.delete("/layers/{layer}/material-combos/{key}")
def delete_matset(
layer: Layer,
key: str,
판: str = "",
owner: str | None = None,
session: Session = Depends(verify_session),
):
return _call(_root(session, layer, owner, write=True), matset.delete, key, 판)
@@ -1,7 +1,7 @@
"""M02 양식 층 API — 계약 `tmp/M02_분석/6_계약.md` 「층 (sub4)」.
⚠ 등록은 `main.py`(로그인만) · 층마다 권한은 여기서:
system 읽기만(고치기는 `/api/m02/templates` 시스템 관리자 길)
system 읽기 · 쓰기는 시스템 관리자(옛 `/api/m02/templates` 길과 같은 저장소)
company 같은 회사 읽기 · 쓰기는 회사 관리자(ADMIN · 마스터 · 시스템 관리자)
personal 본인 읽기·쓰기 · 같은 회사 사람 것은 읽기만(가져오기)
project 같은 회사 프로젝트 · 작업본 쓰기 · `_initial/` 은 안 씀
@@ -26,6 +26,8 @@ from common_util import common_util_spreadsheet as spreadsheet
from common_util.common_util_auth import verify_session
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool, run_with_connection
from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Table_Fill as fill
from M02_MasterTemplete import M02_Template_Layers as layers
@@ -152,8 +154,8 @@ async def _project(session: dict[str, Any], project_id: str | None) -> dict[str,
def _company_of(session: dict[str, Any], project: dict[str, Any] | None) -> int:
company_id = project["company_id"] if project else session.get("company_id")
if company_id is None:
raise HTTPException(status_code=403, detail="회사 연결이 필요합니다.")
if company_id is None: # 회사 없는 사용자는 없음(21-0) — 회사 · 개인 층은 회사부터
raise HTTPException(status_code=409, detail="회사부터 등록 — 회사 연결이 필요합니다.")
return int(company_id)
@@ -174,8 +176,8 @@ async def _layer_dir(
) -> Path:
"""층 폴더 — 읽기·쓰기 권한까지 여기서 가름."""
if layer == "system":
if write:
raise HTTPException(status_code=403, detail="시스템 양식은 마스터 템플릿 화면에서만.")
if write and not _is_system_admin(session):
raise HTTPException(status_code=403, detail="시스템 양식은 시스템 관리자만.")
return layers.system_dir()
if layer == "project":
return layers.project_dir((await _project(session, project_id))["root"])
@@ -184,16 +186,22 @@ async def _layer_dir(
if layer == "company":
if write and not _is_company_admin(session):
raise HTTPException(status_code=403, detail="회사 공식 양식은 회사 관리자만.")
return layers.company_dir(company_id)
return _ensured(layers.company_dir(company_id))
if layer == "personal":
owner = int(user_id) if user_id is not None else int(session["user_id"])
if write and owner != session.get("user_id"):
raise HTTPException(status_code=403, detail="개인 양식은 본인만 고칩니다.")
await _same_company_user(session, owner, company_id)
return layers.personal_dir(company_id, owner)
return _ensured(layers.personal_dir(company_id, owner))
raise HTTPException(status_code=404, detail="없는 층입니다.")
def _ensured(folder: Path) -> Path:
"""회사 · 개인 층 첫 진입 — 빈 구조물집계표 뼈대(PLAN 21-2 · `Items.ensure`)."""
items_store.ensure(folder)
return folder
def _bad(error: ValueError) -> HTTPException:
return HTTPException(status_code=400, detail=str(error))
@@ -261,6 +269,13 @@ async def save_layer_template(
if kind == "table" and fill.is_fillable(document):
# 채운 표가 와도 설계값은 안 받아 적음 — 양식 + 손 값만(5장 · 브라우저 값을 믿지 않음)
document = fill.strip_design(document)
if layer != "project":
# 시스템 · 회사 · 개인 층 = 마스터 저장소 규칙 그대로(집계표 저장 때 새 열 파일 ·
# 번호 · SS 장부 ·
# 계산값 서버 풀이 · 번호 칸 보존 — `M02_MasterTemplete_Store.write`)
return await asyncio.to_thread(
_store_save, folder, kind, name, body.판 or "", document, layer
)
if kind in ("area", "material"):
# 면적산출 · 재료산출 계산값은 서버가 같은 TS 엔진을 Node 로 돌려 새로 적음
# (브라우저 값 버림 · 5장 ②)
@@ -287,6 +302,15 @@ async def save_layer_template(
return {"종류": kind, "이름": name, "판": version, "층": layer}
def _store_save(folder: Path, kind: str, name: str, version: str, doc: Any, layer: str) -> dict:
with store.at(folder):
try:
info = store.write(kind, name, version, doc)
except store.StoreError as error:
raise HTTPException(status_code=error.status, detail=error.detail) from error
return {"종류": kind, "이름": name, "판": info["판"], "층": layer}
@router.delete("/layers/{layer}/templates/{kind}/{name}")
async def delete_layer_template(
layer: Layer,
@@ -296,10 +320,15 @@ async def delete_layer_template(
project_id: str | None = Query(None),
session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
"""개인 = 본인만 · 회사 = 회사 관리자만 · 시스템 · 프로젝트 층은 거절 · 판이 다르면 409."""
if layer in ("system", "project"):
raise HTTPException(status_code=403, detail="개인 · 회사 층 양식만 지울 수 있습니다.")
"""개인 = 본인만 · 회사 = 회사 관리자만 · 시스템 = 시스템 관리자 · 프로젝트 층은 거절 ·
판이 다르면 409 · 프로그램이 이름으로 부르는 양식(`store.IN_USE` — 구조물집계표 등)은 409."""
if layer == "project":
raise HTTPException(
status_code=403, detail="개인 · 회사 · 시스템 층 양식만 지울 수 있습니다."
)
folder = await _layer_dir(session, layer, project_id=project_id, write=True)
if (kind, name) in store.IN_USE:
raise HTTPException(status_code=409, detail=f"쓰는 양식이라 못 지움 「{name}」")
try:
layers.check_kind(kind)
layers.check_name(name)
+34 -11
View File
@@ -1,4 +1,4 @@
"""M02 마스터 템플릿 — 시스템 층 저장소 (`resources/master_template/{종류}/<이름>.json`).
"""M02 마스터 템플릿 — 층 저장소 · 기본 시스템 층(`resources/master_template/{종류}/<이름>.json`).
M01 `Store` 방식 — 판(파일 sha256 앞 16자) · 판이 다르면 409 · 원자 쓰기.
권한은 등록하는 쪽(`main.py`)이 붙임.
@@ -10,6 +10,9 @@ import hashlib
import json
import re
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from datetime import datetime
from pathlib import Path
from typing import Any
@@ -20,6 +23,26 @@ from M02_MasterTemplete import M02_Template_Layers as layers
FOLDER: Path = Path(__file__).resolve().parent.parent / "resources" / "master_template"
KINDS = layers.KINDS # 시험은 FOLDER 를 사본으로 바꿈
# 지금 다루는 층 폴더 — 층 라우터가 `at(층 폴더)` 로 바꿈 · 안 바꾸면 시스템 층(`FOLDER`).
# 회사 · 개인 층도 시스템과 같은 모양(`table/구조물집계표.json` · 조합 장부 ·
# structure/area/material · 자재조합.json — PLAN 21-1)이라 아래 함수는 그대로 씀
_ROOT: ContextVar[Path | None] = ContextVar("m02_layer_root", default=None)
def root() -> Path:
return _ROOT.get() or FOLDER
@contextmanager
def at(folder: str | Path) -> Iterator[Path]:
"""이 안에서 부르는 저장소 함수는 `folder` 층을 봄(요청 하나 · 스레드 하나 안에서만)."""
token = _ROOT.set(Path(folder))
try:
yield Path(folder)
finally:
_ROOT.reset(token)
# 구조물집계표 구조물 열(꼴=수) 마다 세 파일(파일 이름 = 열 id) —
# 구조물 도면 `structure/` (도각 없는 전용 화면) · 면적산출 `area/` · 재료산출 `material/`
# (둘 다 스프레드시트 통합문서)
@@ -75,7 +98,7 @@ def _path(kind: str, name: str) -> Path:
or _BAD_NAME.search(name)
):
raise StoreError(400, f"쓸 수 없는 글자가 있는 이름 「{name}」")
return FOLDER / kind / f"{name}.json"
return root() / kind / f"{name}.json"
def _info(kind: str, path: Path) -> dict[str, Any]:
@@ -91,7 +114,7 @@ def list_all() -> list[dict[str, Any]]:
rows = [
_info(kind, p)
for kind in KINDS
for p in sorted((FOLDER / kind).glob("*.json"))
for p in sorted((root() / kind).glob("*.json"))
if not p.name.startswith(".")
]
return rows
@@ -162,7 +185,7 @@ def delete(kind: str, name: str, version: str | None = None) -> None:
def _docs(kind: str) -> list[tuple[str, Any]]:
folder = FOLDER / kind
folder = root() / kind
return [(p.stem, json.loads(p.read_bytes())) for p in sorted(folder.glob("[!.]*.json"))]
@@ -260,7 +283,7 @@ def _new_code(column: str) -> str:
def _retired() -> list[str]:
"""지운 열의 8자리 — 조합 장부 `지운번호`(한 번 준 번호는 다시 안 줌)."""
path = FOLDER / COMBO_FILE
path = root() / COMBO_FILE
return json.loads(path.read_bytes()).get("지운번호", []) if path.is_file() else []
@@ -318,7 +341,7 @@ def _take_origins(doc: Any) -> dict[str, str]:
source = col.pop("본뜬열", None)
if not source or source not in by_id or not col.get("id"):
continue
if (FOLDER / "structure" / f"{col['id']}.json").is_file():
if (root() / "structure" / f"{col['id']}.json").is_file():
continue
out[col["id"]] = source
for field in ("일위대가", "자재조합"):
@@ -360,14 +383,14 @@ def prune_removed_structures() -> list[str]:
codes: set[str] = set()
with _LOCK:
for kind, key, _, _ in PARTS:
for path in sorted((FOLDER / kind).glob("[!.]*.json")):
for path in sorted((root() / kind).glob("[!.]*.json")):
if path.stem not in keep:
code = _code(json.loads(path.read_bytes()), key)
codes.update([code] if code else [])
path.unlink()
gone.append(f"{kind}/{path.stem}")
if codes: # 지운 번호는 장부에 남겨 새 열이 다시 안 받게
book = FOLDER / COMBO_FILE
book = root() / COMBO_FILE
doc = json.loads(book.read_bytes()) if book.is_file() else {}
doc["지운번호"] = sorted(codes | set(doc.get("지운번호", [])))
atomic_write_json(book, doc)
@@ -377,7 +400,7 @@ def prune_removed_structures() -> list[str]:
def _move_basis() -> list[str]:
"""12장 산출근거 `basis/` → 면적산출 `area/` (내용 · 알약 연결 그대로 · 표번 칸 → 면적번 ·
시트 이름 산출근거 → 면적산출) · 옮긴 뒤 빈 `basis/` 폴더를 지움."""
old = FOLDER / "basis"
old = root() / "basis"
if not old.is_dir():
return []
moved: list[str] = []
@@ -478,7 +501,7 @@ def _combo_row(key: str, col: dict[str, Any], numbers: dict[str, str]) -> dict[s
def sync_combos() -> dict[str, Any]:
"""조합 표를 집계표에 맞춤 — 새 열 = 새 SS 키 · 지운 열 = 줄 지움 · 고친 열 = 칸 갱신 ·
줄 차례 = 집계표 열 차례. 바뀐 것이 없으면 파일을 안 씀."""
path = FOLDER / COMBO_FILE
path = root() / COMBO_FILE
with _LOCK:
old = json.loads(path.read_bytes()) if path.is_file() else {}
keys = {row["열"]: row["키"] for row in old.get("줄", [])}
@@ -502,5 +525,5 @@ def sync_combos() -> dict[str, Any]:
def read_combos() -> dict[str, Any]:
"""조합 표 — 파일이 아직 없으면 지금 열 전부로 처음 한 번 만듦."""
path = FOLDER / COMBO_FILE
path = root() / COMBO_FILE
return json.loads(path.read_bytes()) if path.is_file() else sync_combos()
+25
View File
@@ -0,0 +1,25 @@
/* =============================================================================
* common_util_station.ts
* 측점 화면 표기 한 벌 — B05 · B06 목록 · 폼 · 툴팁 · 종단 표시가 모두 이것을 씀
* (`common_util_svg` 가 다시 내보냄 · B05 는 `formatStation` 이름으로 가져다 씀). 가져오는 것이 없는
* 순수 함수라 서버 Node · 시험 헬퍼에서도 그대로 돎.
* ========================================================================== */
/** 측점 라벨 — "STA" 접두사 없이 `측점번호+잔여거리`(예: `4+0` · `5+20`).
* 화면 표기만 거리를 정수로(119.8 → 5+20 · 2026-09-29 사용자 확정) — 입력 · 저장값은 소수 그대로. */
export function stationLabel(chainage: number, interval: number): string {
const safeInterval = interval > 0 ? interval : 1;
let stationNumber = Math.floor((chainage + 1e-6) / safeInterval);
let remainder = chainage - stationNumber * safeInterval;
if (Math.abs(remainder) < 0.05) remainder = 0;
if (remainder >= safeInterval - 0.05) {
stationNumber += 1;
remainder = 0;
}
return stationPartsLabel(stationNumber, remainder);
}
/** 이미 나눈 측점번호 · 잔여거리의 표기 — 거리는 정수로 보이기만. */
export function stationPartsLabel(stationNumber: number, remainder: number): string {
return `${stationNumber}+${Math.round(remainder)}`;
}
+1 -12
View File
@@ -49,15 +49,4 @@ export function inferStationInterval(stations: Array<{ chainage_m: number }>): n
);
}
/** 측점 라벨 — "STA" 접두사 없이 `측점번호+잔여거리`(예: `4+0.0`). */
export function stationLabel(chainage: number, interval: number): string {
const safeInterval = interval > 0 ? interval : 1;
let stationNumber = Math.floor((chainage + 1e-6) / safeInterval);
let remainder = chainage - stationNumber * safeInterval;
if (Math.abs(remainder) < 0.05) remainder = 0;
if (remainder >= safeInterval - 0.05) {
stationNumber += 1;
remainder = 0;
}
return `${stationNumber}+${remainder.toFixed(1)}`;
}
export { stationLabel } from "./common_util_station";
+2
View File
@@ -63,6 +63,7 @@ from B07_DesignDetail.B07_DesignDetail_Router_Frame import router as b07_frame_r
from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router
from M02_MasterTemplete.M02_MasterTemplete_Router_Drawing import router as m02_drawing_router
from M02_MasterTemplete.M02_MasterTemplete_Router_Items import router as m02_items_router
from M02_MasterTemplete.M02_MasterTemplete_Router_Layers import router as m02_layers_router
from common_util.common_util_audit import note_api_call, record_call_burst
from common_util.common_util_auth import (
@@ -638,6 +639,7 @@ app.include_router(m01_master_data_router, dependencies=system_admin_only)
# 마스터 템플릿 — 시스템 층은 시스템 관리자만 · 층(회사·개인·프로젝트)·도면 길은 로그인만(권한은 길 안에서).
app.include_router(m02_master_template_router, dependencies=system_admin_only)
app.include_router(m02_layers_router, dependencies=protected)
app.include_router(m02_items_router, dependencies=protected)
app.include_router(m02_drawing_router, dependencies=protected)
# 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근).
# 그 위에 서버가 환경까지 한 번 더 본다.
@@ -101,4 +101,16 @@ out.moved = store.edits();
store.resetStation(200);
out.deleted = store.edits();
// 화면 표기 한 벌 — 거리는 정수로 보이기만(PLAN 20장 · 119.8 → 5+20)
const { stationLabel, stationPartsLabel } = loadTs(
path.join(ROOT, "common_util", "common_util_station.ts"),
);
out.labels = [
stationLabel(119.8, INTERVAL),
stationLabel(120, INTERVAL),
stationLabel(84.3, INTERVAL),
stationLabel(123.37, INTERVAL),
stationPartsLabel(5, 19.8),
];
process.stdout.write(JSON.stringify(out));
@@ -22,6 +22,8 @@ from typing import Any
import pytest
from B05_Profile.B05_Profile_Engine_Sections import cross_filename
ROOT = Path(__file__).resolve().parents[2]
PROJECT_ID = "936be972-11bc-46c2-8bf3-b15d8de7df0d" # OWNERS.md — laptop-sub `랩탑_보조`
SOURCE = ROOT / "storage" / "1" / "3" / PROJECT_ID
@@ -425,6 +427,10 @@ class Flow:
near = [s["chainage_m"] for s in lon["stations"] if abs(s["chainage_m"] - c) < 1.0]
if not any(abs(n - c) < 0.005 for n in near):
d.append(f"관 측점 {c} — 종단 측점에 없음(가까운 측점 {near})")
# 소수 측점 횡단(PLAN 20장) — 그 자리 횡단 파일이 소수 2자리 이름으로 따로 있음
name = cross_filename(c)
if not (root / "B06_Section" / "cross_sections" / name).is_file():
d.append(f"관 측점 {c} — 횡단 파일 {name} 없음")
self._check_designs()
def _check_designs(self) -> None:
@@ -192,7 +192,7 @@ def test_matching_signature_loads_points(tmp_path):
]
# ── 횡단 파일명 충돌 — 한 정수 미터에 측점 하나 · 관 · 구조물 자체 측점 우선(PLAN 19장) ──
# ── 소수 측점 횡단 — 관 · 구조물 자리마다 따로(이름 소수 2자리 · PLAN 20장) ──────────────
def _stations(extra_stations):
@@ -204,31 +204,32 @@ def _stations(extra_stations):
return result["longitudinal"]["stations"]
def test_extra_station_in_grid_meter_replaces_grid_station():
def test_extra_station_in_grid_meter_keeps_both_with_own_names():
stations = _stations(((100.4, "X"),))
chainages = [station["chainage_m"] for station in stations]
# 관 · 구조물 자체 측점 그대로 — 같은 정수 미터 격자 100 은 빠져 파일(cross_00100m)은 한 장
assert 100.4 in chainages and 100.0 not in chainages
assert 100.4 in chainages and 100.0 in chainages
labelled = [station for station in stations if station.get("structure") == "X"]
assert [(station["chainage_m"], station["kind"]) for station in labelled] == [
(100.4, "irregular")
]
assert cross_filename(100.4) == "cross_00100.40m.json"
assert len({cross_filename(c) for c in chainages}) == len(chainages)
def test_extra_station_at_route_end_keeps_end_station():
stations = _stations(((0.3, "S"), (349.8, "E")))
chainages = [station["chainage_m"] for station in stations]
# 시 · 종점은 격자 그대로 두고 라벨만 얹음
assert chainages[0] == 0.0 and chainages[-1] == 350.0
assert 0.3 not in chainages and 349.8 not in chainages
assert stations[0].get("structure") == "S" and stations[-1].get("structure") == "E"
def test_extra_station_on_grid_within_5mm_labels_grid_station():
stations = _stations(((100.003, "X"), (0.3, "S")))
labelled = {station["chainage_m"]: station for station in stations if station.get("structure")}
assert sorted(labelled) == [0.3, 100.0]
assert labelled[100.0]["kind"] == "regular" and labelled[0.3]["kind"] == "irregular"
def test_two_extras_in_same_meter_bucket_keep_one():
def test_two_extras_in_same_meter_both_kept():
stations = _stations(((80.2, "A"), (80.4, "B")))
assert [station["chainage_m"] for station in stations if station.get("structure")] == [80.2]
assert 80.0 not in [station["chainage_m"] for station in stations]
assert [station["chainage_m"] for station in stations if station.get("structure")] == [
80.2,
80.4,
]
assert 80.0 in [station["chainage_m"] for station in stations]
def test_live_extras_stay_irregular_and_filenames_stay_unique():
@@ -287,7 +288,7 @@ def test_run_section_generation_derives_stations_from_truth_files(project):
]
cross_dir = project / "B06_Section" / "cross_sections"
for name in ("cross_00084m", "cross_00150m", "cross_00201m", "cross_00276m"):
for name in ("cross_00084.30m", "cross_00149.73m", "cross_00200.92m", "cross_00275.71m"):
assert (cross_dir / f"{name}.json").is_file()
# 파일 = DB 행 — 종전에는 파일 23 vs DB 19로 갈렸다.
assert len(result["cross_sections"]) == len(stations)
+30 -3
View File
@@ -37,6 +37,10 @@ def test_칸을_안_고치면_실어_둔_측점_그대로(results):
assert results["r119"] == ["5", "19.8", 119.8]
def test_화면_표기는_거리를_정수로(results):
assert results["labels"] == ["5+20", "6+0", "4+4", "6+3", "5+20"]
def test_칸을_고치면_친_값(results):
assert results["edited"] == pytest.approx(125.2)
assert results["normalized"] == ["2", "10.0", 50]
@@ -96,8 +100,31 @@ def test_사용자_구조물_119_8_저장_다시_읽기_그대로(tmp_path):
SectionGenerationOptions(extra_stations=extras),
)
chainages = [round(s["chainage_m"], 3) for s in sections["longitudinal"]["stations"]]
# 횡단 측점도 자체 측점 119.8(같은 정수 미터 격자 120 대신 · 파일 이름 cross_00120m 한 장)
assert 119.8 in chainages and 120.0 not in chainages
assert cross_filename(119.8) == cross_filename(120.0)
# 소수 측점 횡단을 따로 둠(PLAN 20장) — 119.8 · 격자 120 둘 다 · 이름도 따로
assert 119.8 in chainages and 120.0 in chainages
assert (cross_filename(119.8), cross_filename(120.0)) == (
"cross_00119.80m.json",
"cross_00120m.json",
)
_, again = load_structures(str(tmp_path))
assert [s.chainage_m for s in again] == [119.8] and again[0].memo == "메모만"
def test_옛_프로젝트_정수_이름_횡단은_찾고_안_지움(tmp_path):
"""소수 측점이 정수 m 이름으로 저장된 옛 프로젝트 — 열 때 지우지 않고 그 파일을 씀."""
from B05_Profile.B05_Profile_Engine_Sections import find_cross_file, prune_stale_cross_files
(tmp_path / "cross_00084m.json").write_text("{}", encoding="utf-8") # 옛 84.3
(tmp_path / "cross_00100m.json").write_text("{}", encoding="utf-8")
(tmp_path / "cross_00119.80m.json").write_text("{}", encoding="utf-8") # 새 이름
(tmp_path / "cross_00500m.json").write_text("{}", encoding="utf-8") # 잔재
stations = [{"chainage_m": 84.3}, {"chainage_m": 100.0}, {"chainage_m": 119.8}]
valid = prune_stale_cross_files(tmp_path, stations)
assert sorted(p.name for p in tmp_path.iterdir()) == [
"cross_00084m.json",
"cross_00100m.json",
"cross_00119.80m.json",
]
assert valid == {"cross_00084m.json", "cross_00100m.json", "cross_00119.80m.json"}
assert find_cross_file(tmp_path, 84.3).name == "cross_00084m.json"
assert find_cross_file(tmp_path, 119.8).name == "cross_00119.80m.json"
@@ -34,3 +34,9 @@ def test_행_없는_측점_수정이면_측점을_세움(monkeypatch):
def test_모든_수정에_행이_있으면_안_부름(monkeypatch):
assert _calls(monkeypatch, [120.0, 140.004], [120.0, 140.0]) == []
assert _calls(monkeypatch, [], [120.0]) == []
def test_소수_측점_수정은_정수_격자_행으로_갈음하지_않음(monkeypatch):
# 119.8 은 격자 120 과 다른 측점(횡단 cross_00119.80m · PLAN 20장) — 제 행이 없으면 세움
assert _calls(monkeypatch, [119.8], [120.0]) == [PROJECT]
assert _calls(monkeypatch, [119.8], [119.8, 120.0]) == []
+172
View File
@@ -0,0 +1,172 @@
"""M02 층 집계표(PLAN 21-2) — 층별 읽기 · 쓰기 · 권한 · 첫 진입 뼈대 · 항목 목록 · 항목 복사 ·
시스템 회귀.
임시 storage(회사 · 개인 층) + 견본 시스템 집계표(`write_sample_table`)로만 봄 —
실제 양식은 안 읽음(시스템 층 두 자리 `store.FOLDER` · `layers.SYSTEM_ROOT` 모두 사본).
자재 코드는 실제 M01 마스터에 있는 MT 키를 씀(자재 조합 저장이 마스터 대조를 함).
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from common_util.common_util_auth import verify_session
from config import config_system
from M01_MasterData import M01_MasterData_Store as m01
from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layers_router
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers
from resources.tester.test_m02_structure import write_sample_table
TABLE = "table/구조물집계표"
def _material() -> dict[str, Any]:
for data in m01.loaded()[0].values():
for row in data.get("줄") or []:
if str(row.get("키", "")).startswith("MT"):
return {"코드": row["키"], "이름": row.get("이름"), "규격": None, "단위": None}
raise AssertionError("마스터에 자재 키가 없음")
@pytest.fixture()
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
system = tmp_path / "system"
system.mkdir()
write_sample_table(system)
monkeypatch.setattr(store, "FOLDER", system)
monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
store.renumber()
storage = tmp_path / "storage"
storage.mkdir()
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
session: dict[str, Any] = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
app = FastAPI()
app.include_router(layers_router.router)
app.include_router(items_router.router)
app.dependency_overrides[verify_session] = lambda: session
return {"client": TestClient(app), "session": session, "storage": storage, "system": system}
def _as(world: dict, role: str = "USER", user: int = 42, company: int | None = 7) -> TestClient:
world["session"].update(role=role, user_id=user, company_id=company)
return world["client"]
def _add_column(client: TestClient, layer: str, col: dict[str, Any]) -> None:
url = f"/api/m02/layers/{layer}/templates/{TABLE}"
got = client.get(url).json()
got["문서"]["열"].append(col)
put = client.put(url, json={"판": got["판"], "문서": got["문서"]})
assert put.status_code == 200, put.text
COL = {"id": "cw", "머리": ["돌기슭막이", "찰쌓기", "회사형", None], "단위": "m", "꼴": "수"}
def test_시스템_층은_그대로_항목_목록은_SS_줄(world):
client = _as(world)
rows = client.get("/api/m02/items").json()
ss = store.read_combos()["줄"]
assert [r["key"] for r in rows] == [r["키"] for r in ss] and rows[0]["key"] == "SS000001"
assert set(rows[0]) == {"key", *items_router.items_store.ITEM_FIELDS}
assert client.get("/api/m02/layers/system/structure-combos").json() == store.read_combos()
# 시스템 층 쓰기는 SYSTEM_ADMIN 만
assert client.post("/api/m02/layers/system/structures/rv").status_code == 403
url = f"/api/m02/layers/system/templates/{TABLE}"
got = client.get(url).json()
assert client.put(url, json={"판": got["판"], "문서": got["문서"]}).status_code == 403
assert client.delete(f"{url}?판={got['판']}").status_code == 403
admin = _as(world, "SYSTEM_ADMIN")
assert admin.put(url, json={"판": got["판"], "문서": got["문서"]}).status_code == 200
assert admin.delete(f"{url}?판={got['판']}").status_code == 409 # 쓰는 양식
def test_회사_층_첫_진입_뼈대와_쓰기_권한(world):
client = _as(world)
assert client.get("/api/m02/items?layer=company").json() == []
root = world["storage"] / "7" / "templates"
table = json.loads((root / "table/구조물집계표.json").read_text("utf-8"))
assert (
table["열"] == []
and table["층"]
== json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8"))["층"]
)
assert json.loads((root / "구조물집계표조합.json").read_text("utf-8"))["다음"] == 1
assert json.loads((root / "자재조합.json").read_text("utf-8"))["다음"] == 1
url = f"/api/m02/layers/company/templates/{TABLE}"
got = client.get(url).json()
got["문서"]["열"].append(COL)
assert client.put(url, json={"판": got["판"], "문서": got["문서"]}).status_code == 403
_add_column(_as(world, "ADMIN"), "company", COL)
items = _as(world).get("/api/m02/items?layer=company").json() # 같은 회사 일반 사용자
assert [(r["key"], r["열"]) for r in items] == [("SS000001", "cw")]
assert items[0]["도번"].startswith("SD") and (root / "structure/cw.json").is_file()
assert client.get("/api/m02/items?layer=company&owner=9").status_code == 403 # 남의 회사
# 일위대가(UA) 갈라 내기는 M01 시스템 마스터에 쓰므로 회사 관리자도 못 함
body = {"본뜬키": "UA000001", "담은로직": [], "이름": "x"}
admin = _as(world, "ADMIN")
assert admin.post("/api/m02/layers/company/structures/cw/combo", json=body).status_code == 403
assert admin.delete(f"{url}?판={client.get(url).json()['판']}").status_code == 409 # 쓰는 양식
def test_개인_층은_본인만_회사_없으면_409(world):
client = _as(world)
_add_column(client, "personal", {**COL, "id": "p1"})
assert [r["열"] for r in client.get("/api/m02/items?layer=personal").json()] == ["p1"]
assert (world["storage"] / "7" / "42" / "templates" / "area" / "p1.json").is_file()
other = client.get("/api/m02/items?layer=personal&owner=7/43")
assert other.status_code == 403
lonely = _as(world, company=None)
groups = lonely.get("/api/m02/items?all=1").json()
assert [(g["layer"], g["items"]) for g in groups][1:] == [("company", []), ("personal", [])]
assert len(groups[0]["items"]) == len(store.read_combos()["줄"]) # 마스터는 보임
made = lonely.post(
"/api/m02/layers/personal/material-combos", json={"줄": {"이름": "a", "담은자재": []}}
)
assert made.status_code == 409 and "회사부터" in made.json()["detail"]
def test_회사_항목을_개인으로_복사(world):
admin = _as(world, "ADMIN")
_add_column(admin, "company", COL)
base = "/api/m02/layers/company"
drawing = admin.get(f"{base}/templates/structure/cw").json()
doc = {**drawing["문서"], "도면": {**drawing["문서"]["도면"], "entities": [{"type": "line"}]}}
admin.put(f"{base}/templates/structure/cw", json={"판": drawing["판"], "문서": doc})
판 = admin.get(f"{base}/material-combos").json()["판"]
row = {"이름": "회사 자재", "담은자재": [_material()]}
um = admin.post(f"{base}/material-combos", json={"판": 판, "줄": row}).json()["키"]
admin.post(f"{base}/structures/cw/use-combo", json={"종류": "자재조합", "키": um})
ss = admin.get("/api/m02/items?layer=company").json()[0]["key"]
user = _as(world)
body = {"from_layer": "company", "from_owner": "7", "key": ss}
got = user.post("/api/m02/layers/personal/items/copy", json=body)
assert got.status_code == 200, got.text
made = got.json()
assert made["열"] == "cw" and made["key"] == "SS000001"
mine = "/api/m02/layers/personal"
copy = user.get(f"{mine}/templates/structure/cw").json()["문서"]
assert copy["도면"]["entities"] == [{"type": "line"}] and copy["열"] == "cw"
col = next(c for c in user.get(f"{mine}/templates/{TABLE}").json()["문서"]["열"])
assert col["본뜬"] == {"layer": "company", "owner": "7", "key": ss}
mat = user.get(f"{mine}/material-combos/{col['자재조합']}").json()
assert mat["이름"] == "회사 자재" and mat["담은자재"][0]["코드"] == _material()["코드"]
# 또 복사하면 새 열 id · 새 SS 키 · 자재 조합 이름 (2)
again = user.post("/api/m02/layers/personal/items/copy", json=body).json()
assert (again["열"], again["key"]) == ("cw_2", "SS000002")
names = [r["이름"] for r in user.get(f"{mine}/material-combos").json()["줄"]]
assert names == ["회사 자재", "회사 자재 (2)"]
# 마스터 항목도 개인으로 · 없는 키 404
master = {"from_layer": "system", "key": "SS000001"}
assert user.post("/api/m02/layers/personal/items/copy", json=master).status_code == 200
missing = {**body, "key": "SS999999"}
assert user.post("/api/m02/layers/personal/items/copy", json=missing).status_code == 404
+3 -2
View File
@@ -143,12 +143,13 @@ def test_개인_양식은_본인만_쓰고_같은_회사는_읽어_가져옴(wor
session.update(user_id=43)
theirs = mine + "&user_id=42"
assert client.get(theirs).status_code == 200
# 첫 진입 때 구조물집계표 빈 뼈대가 서므로(PLAN 21-2) 새 이름으로 씀
put = client.put(
f"/api/m02/layers/personal/templates/table/구조물집계표?project_id={P1}&user_id=42",
f"/api/m02/layers/personal/templates/table/내표?project_id={P1}&user_id=42",
json={"판": None, "문서": {"열": []}},
)
assert put.status_code == 200 # user_id 는 쓰기에서 무시 — 본인(43) 자리에 새로 씀
assert (world["storage"] / "7/43/templates/table/구조물집계표.json").is_file()
assert (world["storage"] / "7/43/templates/table/내표.json").is_file()
got = client.post(
f"/api/m02/projects/{P2}/templates/apply", json={"from": "personal", "user_id": 42}
)