Merge remote-tracking branch 'origin/sub_laptop_4' into sub_laptop_1

B02_ProjRegister/B02_ProjRegister_Repository.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from datetime import datetime
 from pathlib import Path
 from typing import Any
@@ -16,6 +17,9 @@ from common_util.common_util_workflow import load_project_workflow
 from common_util.common_util_workflow_state import initialize_project_stages
 from config.config_db import get_db_pool
 from config.config_system import STORAGE_BASE_DIR
+from M02_MasterTemplete.M02_Template_Layers import seed_project
+
+logger = logging.getLogger(__name__)
 
 
 def _build_project_storage(company_id: int, user_id: int, project_id: str) -> tuple[str, Path]:
@@ -40,6 +44,12 @@ def _initialize_project_storage(project_root: Path, project_id: str) -> None:
             "stages": [f"{stage}/{subdir}" for stage, subdir in PROJECT_STORAGE_LAYOUT_V2],
         },
     )
+    # 마스터 템플릿 — 시스템 양식 전부를 작업본 + `_initial/` 로(회사 공식이 있어도 시스템).
+    # 복사가 실패해도 프로젝트는 만듦 — 읽는 쪽이 시스템 양식으로 떨어짐.
+    try:
+        seed_project(project_root)
+    except OSError:
+        logger.exception("프로젝트 양식 복사 실패: project_id=%s", project_id)
 
 
 async def create_project(

M02_MasterTemplete/M02_Table_Fill.py

@@ -0,0 +1,464 @@
+"""M02 표 양식 설계값 채우기 — 서버 단독(CLAUDE.md 5장 ③) · 저장 안 함.
+
+프로젝트 설계 정본 셋을 읽어 표 문서의 **바인딩 열**을 채움:
+    `B05_Profile/route/structures.json`   놓인 구조물
+    `.../drainage/edits/pipe_points.json` 계곡 통과 시설(관 · BOX · 물넘이 · 세월교 · 기슭막이)
+    B06 횡단 설계 `design.pipe_length_m`  관 연장(부르는 쪽이 DB 에서 읽어 넘김)
+
+- 펼침 열 — 마스터 한 칸 → 설계에 쓰인 값마다 열(choices 차례) · 쓰인 값이 없으면 열을 뺌.
+  열 id = `마스터 id|펼친 값|…`(머리 글자가 바뀌어도 id 는 그대로).
+- 줄 — 측점마다 한 줄(같은 측점 구조물은 한 줄) · 점 `NO.x+y` · 구간 `NO.a~NO.b` ·
+  같은 측점에 관이 둘이면 줄을 나눔 · 줄 id 는 측점에서 지어 손 값이 다시 채워도 붙어 있음.
+- 손 열 값 · 전구간 줄 · 사용자가 더한 줄(`손`)은 작업본 것을 그대로 지킴.
+- 값을 여기서 짓지 않음 — 정본 값을 모으기만. 빈 값은 빈칸(0 으로 안 채움).
+"""
+
+from __future__ import annotations
+
+import copy
+import re
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+from B05_Profile.B05_Profile_Structures_Repository import load_structures
+from B05_Profile.B05_Profile_Structures_Schema import StructureType, structure_type_map
+from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
+
+#: 관 측점 ↔ 횡단 측점 맞추기 허용(옛 배수관 물량과 같은 규칙) — 넓히면 옆 관 연장을 물어 옴.
+PIPE_MATCH_TOLERANCE_M = 0.5
+#: 계곡 통과 시설 `facility` → 종류(빈 값은 관).
+FACILITY_TYPES = {
+    "pipe": "pipe",
+    "box_culvert": "box_culvert",
+    "ford_pavement": "ford_pavement",
+    "ford_bridge": "ford_bridge",
+    "revetment": "revetment",
+}
+EMPTY_LABEL = "-"
+_REF = re.compile(r"\[([^\]$@][^\]@]*)\]")
+
+
+# ── 설계 읽기 ─────────────────────────────────────────
+
+
+def pipe_lengths_from_designs(designs: Iterable[dict[str, Any]]) -> dict[float, float]:
+    """B06 횡단 설계 `[{chainage_m, design}]` → `{측점: 관 연장}` · 없는 측점은 안 담음."""
+    found: dict[float, float] = {}
+    for row in designs or []:
+        design = row.get("design") if isinstance(row, dict) else None
+        if not isinstance(design, dict):
+            continue
+        length = _number(design.get("pipe_length_m"))
+        if length and length > 0:
+            found[round(float(row.get("chainage_m") or 0.0), 3)] = length
+    return found
+
+
+def _nearest(lengths: dict[float, float], chainage: float) -> float | None:
+    if not lengths:
+        return None
+    key = round(chainage, 3)
+    if key in lengths:
+        return lengths[key]
+    best = min(lengths, key=lambda x: abs(x - chainage))
+    return lengths[best] if abs(best - chainage) <= PIPE_MATCH_TOLERANCE_M else None
+
+
+def _with_defaults(definition: StructureType | None, options: dict[str, Any]) -> dict[str, Any]:
+    """빠진 칸은 레지스트리 기본값 — B05 화면이 보여 주는 값과 같게."""
+    merged = dict(options or {})
+    for field in definition.options if definition else []:
+        if field.key not in merged and field.default is not None:
+            merged[field.key] = field.default
+    return merged
+
+
+def collect_items(
+    project_root: str | Path,
+    pipe_lengths: dict[float, float] | None = None,
+    types: dict[str, StructureType] | None = None,
+) -> list[dict[str, Any]]:
+    """설계 정본 → 구조물 목록.
+
+    한 건 = `{type_id, placement, chainage_m, start_m, end_m, options, 관연장}`.
+    """
+    types = types or structure_type_map()
+    lengths = pipe_lengths or {}
+    items: list[dict[str, Any]] = []
+    _, structures = load_structures(str(project_root))
+    for item in structures:
+        definition = types.get(item.type_id)
+        if definition is not None and definition.managed_by:
+            continue  # 관 지점 정본이 주인 — 옛 저장분이 남아 있어도 두 번 안 셈
+        items.append(
+            {
+                "type_id": item.type_id,
+                "placement": definition.placement if definition else "point",
+                "chainage_m": float(item.chainage_m),
+                "start_m": item.start_m,
+                "end_m": item.end_m,
+                "options": _with_defaults(definition, dict(item.options or {})),
+            }
+        )
+    for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))):
+        type_id = FACILITY_TYPES.get(point.facility or "pipe", point.facility)
+        chainage = float(point.chainage_m)
+        items.append(
+            {
+                "type_id": type_id,
+                "placement": "point",
+                "chainage_m": chainage,
+                "start_m": None,
+                "end_m": None,
+                "options": _with_defaults(types.get(type_id), dict(point.options or {})),
+                "관연장": _nearest(lengths, chainage) if type_id == "pipe" else None,
+            }
+        )
+    return items
+
+
+# ── 값 · 글 ───────────────────────────────────────────
+
+
+def _number(value: Any) -> float | None:
+    if isinstance(value, bool) or value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _label(key: str, value: Any) -> str:
+    """펼친 값 글 — 높이 등 `_m` 칸은 `2.0` 처럼 소수 한 자리는 남김."""
+    if value is None or value == "":
+        return EMPTY_LABEL
+    number = _number(value)
+    if number is None or isinstance(value, str) and not key.endswith("_m"):
+        return str(value)
+    if number.is_integer():
+        return f"{number:.1f}" if key.endswith("_m") else str(int(number))
+    return f"{number:.3f}".rstrip("0").rstrip(".")
+
+
+def _length(item: dict[str, Any]) -> float | None:
+    stated = _number(item["options"].get("length_m"))
+    if stated:
+        return stated
+    start, end = _number(item.get("start_m")), _number(item.get("end_m"))
+    return abs(end - start) if start is not None and end is not None else None
+
+
+def _source_value(source: dict[str, Any], item: dict[str, Any]) -> float | None:
+    kind = source.get("값")
+    if kind == "개소":
+        return 1.0
+    if kind == "길이":
+        return _length(item)
+    if kind == "관연장":
+        return item.get("관연장")
+    if not kind or kind == "식":
+        return None
+    return _number(item["options"].get(kind))
+
+
+def _matches(source: dict[str, Any], item: dict[str, Any]) -> bool:
+    if source.get("종류") != item["type_id"]:
+        return False
+    for key, want in (source.get("조건") or {}).items():
+        if str(item["options"].get(key)) != str(want):
+            return False
+    # 펼칠 칸이 모두 비면 그 시설이 없는 것(예: 날개벽 형식 없음 → 관보호공 없음)
+    keys = [key for key in source.get("펼침") or [] if not key.startswith("=")]
+    return not keys or any(item["options"].get(key) not in (None, "") for key in keys)
+
+
+def _sources(binding: dict[str, Any]) -> list[dict[str, Any]]:
+    main = {key: binding.get(key) for key in ("종류", "펼침", "값", "조건")}
+    return [main, *(binding.get("더함") or [])]
+
+
+def _expansion(source: dict[str, Any], item: dict[str, Any]) -> tuple[str, ...]:
+    values = []
+    for key in source.get("펼침") or []:
+        values.append(key[1:] if key.startswith("=") else _label(key, item["options"].get(key)))
+    return tuple(values)
+
+
+def _position_orders(binding: dict[str, Any], choices: dict[str, list[str]]) -> list[list[str]]:
+    """펼침 자리마다 차례 목록 — 고정 글(`=찰쌓기`)은 출처 차례 · 칸은 choices 차례."""
+    orders: list[list[str]] = []
+    for source in _sources(binding):
+        for index, key in enumerate(source.get("펼침") or []):
+            while len(orders) <= index:
+                orders.append([])
+            listed = [key[1:]] if key.startswith("=") else choices.get(key, [])
+            orders[index].extend(value for value in listed if value not in orders[index])
+    return orders
+
+
+def _order_key(values: tuple[str, ...], orders: list[list[str]]) -> tuple[Any, ...]:
+    """차례 목록 → 수 차례 → 글 차례 · 빈 값(-)은 끝."""
+    out: list[Any] = []
+    for index, value in enumerate(values):
+        listed = orders[index] if index < len(orders) else []
+        number = _number(value)
+        out.append(
+            (
+                value == EMPTY_LABEL,
+                listed.index(value) if value in listed else len(listed),
+                number if number is not None else float("inf"),
+                value,
+            )
+        )
+    return tuple(out)
+
+
+# ── 측점 ──────────────────────────────────────────────
+
+
+def station_label(chainage_m: float, interval_m: float = 20.0) -> str:
+    """`NO.12` · `NO.12+5` · `NO.12+5.5` — 실무 구조물위치 표기."""
+    safe = interval_m if interval_m > 0 else 20.0
+    number = int((chainage_m + 1e-6) // safe)
+    remainder = round(chainage_m - number * safe, 2)
+    if remainder >= safe - 0.005:
+        number, remainder = number + 1, 0.0
+    if abs(remainder) < 0.005:
+        return f"NO.{number}"
+    text = f"{remainder:.2f}".rstrip("0").rstrip(".")
+    return f"NO.{number}+{text}"
+
+
+def _span(item: dict[str, Any]) -> tuple[float, float | None]:
+    start, end = _number(item.get("start_m")), _number(item.get("end_m"))
+    if item["placement"] == "interval" and start is not None and end is not None:
+        low, high = sorted((start, end))
+        if high - low > 1e-6:
+            return round(low, 2), round(high, 2)
+    return round(float(item["chainage_m"]), 2), None
+
+
+def _row_id(start: float, end: float | None, index: int) -> str:
+    text = f"s{start:.2f}" + (f"~{end:.2f}" if end is not None else "")
+    return text + (f"#{index}" if index else "")
+
+
+# ── 채우기 ────────────────────────────────────────────
+
+
+def _choices(types: dict[str, StructureType]) -> dict[str, list[str]]:
+    found: dict[str, list[str]] = {}
+    for definition in types.values():
+        for field in definition.options:
+            if field.choices and field.key not in found:
+                found[field.key] = list(field.choices)
+    return found
+
+
+def _rewrite(formula: str, siblings: set[str], suffix: str) -> str:
+    return _REF.sub(
+        lambda m: f"[{m.group(1)}|{suffix}]" if m.group(1) in siblings else m.group(0), formula
+    )
+
+
+def _blank_vars(formula: str, variables: dict[str, Any]) -> list[str]:
+    names = re.findall(r"\[\$([^\]]+)\]", formula)
+    return [name for name in names if variables.get(name) in (None, "")]
+
+
+def _unexpand(columns: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """채운 표가 작업본으로 저장됐으면 펼친 열을 마스터 칸(`원열`)으로 되돌림 — 두 번 안 펼침."""
+    out: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for column in columns:
+        origin = (column.get("바인딩") or {}).get("원열")
+        if origin is None:
+            out.append(column)
+        elif origin["id"] not in seen:
+            seen.add(origin["id"])
+            out.append(origin)
+    return out
+
+
+def _expand_columns(
+    columns: list[dict[str, Any]],
+    items: list[dict[str, Any]],
+    variables: dict[str, Any],
+    choices: dict[str, list[str]],
+) -> tuple[list[dict[str, Any]], dict[str, list[tuple[str, str, dict[str, Any]]]]]:
+    """펼침 열 → 값마다 열. 돌려주는 둘째 = `{master id: [(열 id, suffix, 펼친 열)]}`."""
+    columns = _unexpand(columns)
+    groups: dict[str, set[tuple[str, ...]]] = {}
+    orders_of: dict[str, list[list[str]]] = {}
+    for column in columns:
+        binding = column.get("바인딩") or {}
+        if not column.get("펼침") or not binding:
+            continue
+        group = binding.get("묶음") or column["id"]
+        orders_of.setdefault(group, _position_orders(binding, choices))
+        found = groups.setdefault(group, set())
+        for source in _sources(binding):
+            # 값이 빈 구조물도 열은 세움 — 놓였는데 수량이 빈칸인 것이 보여야 함
+            found.update(_expansion(source, item) for item in items if _matches(source, item))
+
+    members: dict[str, set[str]] = {}
+    for column in columns:
+        binding = column.get("바인딩") or {}
+        if column.get("펼침") and binding:
+            members.setdefault(binding.get("묶음") or column["id"], set()).add(column["id"])
+
+    out: list[dict[str, Any]] = []
+    made: dict[str, list[tuple[str, str, dict[str, Any]]]] = {}
+    for column in columns:
+        binding = column.get("바인딩") or {}
+        if not column.get("펼침") or not binding:
+            out.append(copy.deepcopy(column))
+            continue
+        group = binding.get("묶음") or column["id"]
+        ordered = sorted(groups.get(group, set()), key=lambda v: _order_key(v, orders_of[group]))
+        names = binding.get("이름") or {}
+        for values in ordered:
+            shown = [names.get(value, value) for value in values]
+            suffix = "|".join(values)
+            new = copy.deepcopy(column)
+            new["id"] = f"{column['id']}|{suffix}"
+            new["펼침"] = False
+            new["머리"] = [
+                None if part is None else part.format(*shown)
+                for part in (binding.get("머리틀") or column["머리"])
+            ]
+            new["바인딩"] = {**binding, "펼친값": list(values), "원열": copy.deepcopy(column)}
+            formula = binding.get("식틀")
+            formula = formula.format(*values) if formula else column.get("식")
+            if formula:
+                formula = _rewrite(formula, members.get(group, set()), suffix)
+                if _blank_vars(formula, variables):
+                    new.pop("식", None)
+                    new["설명"] = f"{column.get('설명', '')} · 변수가 빈칸이라 계산 안 함".strip()
+                else:
+                    new["식"] = formula
+            out.append(new)
+            made.setdefault(column["id"], []).append((new["id"], suffix, new))
+    return out, made
+
+
+def fill_document(
+    document: dict[str, Any],
+    items: list[dict[str, Any]],
+    types: dict[str, StructureType] | None = None,
+) -> dict[str, Any]:
+    """표 문서 + 설계 구조물 목록 → 채운 표 문서(새 사본)."""
+    types = types or structure_type_map()
+    doc = copy.deepcopy(document)
+    variables = dict(doc.get("변수") or {})
+    interval = _number(variables.get("측점간격_m")) or 20.0
+    columns, made = _expand_columns(doc.get("열") or [], items, variables, _choices(types))
+    notes: list[str] = []
+
+    # 줄 자리 — 관은 한 줄에 하나(둘째 관부터 줄을 나눔)
+    slots: dict[tuple[float, float | None, int], dict[str, Any]] = {}
+    pipes_at: dict[tuple[float, float | None], int] = {}
+    item_slot: list[tuple[float, float | None, int]] = []
+    for item in items:
+        start, end = _span(item)
+        index = 0
+        if item["type_id"] == "pipe":
+            index = pipes_at.get((start, end), 0)
+            pipes_at[(start, end)] = index + 1
+        slot = (start, end, index)
+        slots.setdefault(slot, {"id": _row_id(*slot), "값": {}})
+        item_slot.append(slot)
+
+    for column in columns:
+        binding = column.get("바인딩") or {}
+        if not binding or column.get("손"):
+            continue
+        expanded = binding.get("펼친값")
+        for source in _sources(binding):
+            if source.get("값") in (None, "식"):
+                continue
+            for item, slot in zip(items, item_slot, strict=True):
+                if not _matches(source, item):
+                    continue
+                if expanded is not None and list(_expansion(source, item)) != expanded:
+                    continue
+                value = _source_value(source, item)
+                if value is None:
+                    continue
+                cells = slots[slot]["값"]
+                cells[column["id"]] = round((cells.get(column["id"]) or 0) + value, 3)
+
+    # 포장 줄눈 간격이 변수와 다르면 그 줄 식에 박음(줄 식이 열 식을 이김)
+    base_spacing = _number(variables.get("수축줄눈_간격_m"))
+    for item, slot in zip(items, item_slot, strict=True):
+        spacing = _number(item["options"].get("joint_spacing_m"))
+        if item["type_id"] != "pavement_concrete" or spacing in (None, base_spacing):
+            continue
+        for new_id, _, new in made.get("pv_jt", []):
+            if list(_expansion({"펼침": new["바인딩"]["펼침"]}, item)) == new["바인딩"]["펼친값"]:
+                text = new.get("식", "").replace("[$수축줄눈_간격_m]", _label("", spacing))
+                slots[slot].setdefault("식", {})[new_id] = text
+
+    missing = sum(1 for item in items if item["type_id"] == "pipe" and not item.get("관연장"))
+    if missing:
+        notes.append(f"관 연장 없음 {missing}곳 — B06 횡단 설계 전이면 빈칸")
+    loose = sum(
+        1 for item in items if item["type_id"] == "guardrail" and not item["options"].get("kind")
+    )
+    if loose:
+        notes.append(f"가드레일·경계석·위험표지 종류 미정 {loose}건 — 표에 안 셈")
+
+    # 손 값 지키기 — 같은 줄 id 의 손 열 · 전구간 줄 · 사용자가 더한 줄
+    hand = {column["id"] for column in columns if column.get("손")}
+    old_rows = {row.get("id"): row for row in doc.get("줄") or []}
+    fixed = [row for row in doc.get("줄") or [] if row.get("고정")]
+    added = [row for row in doc.get("줄") or [] if row.get("손") and not row.get("고정")]
+    design_rows = []
+    for number, slot in enumerate(sorted(slots), start=1):
+        row = slots[slot]
+        start, end, _ = slot
+        label = station_label(start, interval)
+        if end is not None:
+            label = f"{label}~{station_label(end, interval)}"
+        row["값"].update({"no": str(number), "sta": label})
+        previous = old_rows.get(row["id"]) or {}
+        for key, value in (previous.get("값") or {}).items():
+            if key in hand:
+                row["값"][key] = value
+        design_rows.append(row)
+    # 설계에서 빠진 줄의 손 값은 버리지 않음 — 사용자 줄로 남겨 보이게
+    kept = {row["id"] for row in design_rows}
+    for row in doc.get("줄") or []:
+        if row.get("고정") or row.get("손") or row.get("id") in kept:
+            continue
+        values = {
+            k: v for k, v in (row.get("값") or {}).items() if k in hand and v not in (None, "")
+        }
+        if values:
+            sta = (row.get("값") or {}).get("sta")
+            added.append({"id": row["id"], "값": {"sta": sta, **values}, "손": True})
+    doc["열"] = columns
+    doc["줄"] = [*fixed, *design_rows, *added]
+    doc["알림"] = notes
+    return doc
+
+
+def fill_table(
+    project_root: str | Path,
+    document: dict[str, Any],
+    pipe_lengths: dict[float, float] | None = None,
+) -> dict[str, Any]:
+    """프로젝트 설계로 표 문서를 채움 — 저장 안 함."""
+    types = structure_type_map()
+    return fill_document(document, collect_items(project_root, pipe_lengths, types), types)
+
+
+def recalc(document: dict[str, Any]) -> dict[str, Any] | None:
+    """계산 열 — 표 재계산(sub3 `common_util_sheet_recalc`)이 있으면 그 결과 · 없으면 None."""
+    try:
+        from common_util.common_util_sheet_recalc import recalc_sheet
+    except ImportError:
+        return None
+    return recalc_sheet(document)

M02_MasterTemplete/M02_Template_Migrate.py

@@ -0,0 +1,119 @@
+"""이미 만든 프로젝트에 시스템 양식 넣기 — 한 번 돌리는 도구 (PLAN 10-5).
+
+프로젝트마다 `seed_project(only_missing=True)` — 작업본 · `_initial/` 에 **없는 것만 더함**.
+이미 있는 양식 · 다른 파일은 안 건드림 · 폴더가 없는 프로젝트는 건너뜀(만들지 않음).
+다시 돌려도 됨 — 시스템 양식이 늘었으면 그 몫만 더해짐.
+
+    ./venv/Scripts/python.exe M02_MasterTemplete/M02_Template_Migrate.py [--dry-run] [--report 파일]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from datetime import datetime
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from config import config_system  # noqa: E402
+from config.config_db import close_db_pool, get_db_pool, init_db_pool  # noqa: E402
+from M02_MasterTemplete import M02_Template_Layers as layers  # noqa: E402
+
+
+async def _projects() -> list[dict[str, Any]]:
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        await cursor.execute(
+            """SELECT p.id, p.name, p.company_id, p.user_id, p.storage_path, u.name
+               FROM projects p LEFT JOIN users u ON u.id = p.user_id
+               WHERE p.deleted_at IS NULL ORDER BY p.company_id, p.created_at"""
+        )
+        rows = await cursor.fetchall()
+    keys = ("id", "name", "company_id", "user_id", "storage_path", "owner")
+    return [dict(zip(keys, row, strict=True)) for row in rows]
+
+
+def project_root(storage_path: str | None) -> Path | None:
+    """DB 저장 경로 → 실경로 · 없거나 수상하면 None(폴더를 만들지 않음)."""
+    if not storage_path:
+        return None
+    parts = PurePosixPath(storage_path.replace("\\", "/")).parts
+    if not parts or parts[0] != "storage" or ".." in parts or len(parts) < 2:
+        return None
+    root = Path(os.path.realpath(config_system.STORAGE_BASE_DIR))
+    path = Path(os.path.realpath(root.joinpath(*parts[1:])))
+    if root not in path.parents or not path.is_dir():
+        return None
+    return path
+
+
+def migrate(projects: list[dict[str, Any]], *, dry_run: bool = False) -> list[dict[str, Any]]:
+    """프로젝트마다 넣은 것 `[{…프로젝트, 작업본, 초기, 상태}]`."""
+    system = {f"{row['종류']}/{row['이름']}" for row in layers.list_templates(layers.system_dir())}
+    report = []
+    for project in projects:
+        root = project_root(project.get("storage_path"))
+        entry = {**project, "작업본": [], "초기": [], "상태": ""}
+        if root is None:
+            entry["상태"] = "폴더 없음 — 건너뜀"
+        elif dry_run:
+            have = {
+                f"{row['종류']}/{row['이름']}"
+                for row in layers.list_templates(layers.project_dir(root))
+            }
+            entry["작업본"] = sorted(system - have)
+            entry["상태"] = "넣을 것(시험)"
+        else:
+            done = layers.seed_project(root, only_missing=True)
+            entry.update(작업본=done["작업본"], 초기=done["초기"])
+            entry["상태"] = "넣음" if done["작업본"] or done["초기"] else "이미 있음"
+        report.append(entry)
+    return report
+
+
+def render(report: list[dict[str, Any]], system: list[dict[str, Any]]) -> str:
+    names = " · ".join(f"{row['종류']}/{row['이름']}" for row in system) or "없음"
+    lines = [
+        "# 기존 프로젝트 양식 넣기",
+        "",
+        f"- 돌린 때: {datetime.now().isoformat(timespec='seconds')}",
+        f"- 시스템 양식: {names}",
+        "- 방법: 작업본 · `_initial/` 에 없는 것만 더함 · 있는 양식 · 다른 파일은 그대로",
+        f"- 프로젝트 {len(report)}개 · 넣음 {sum(r['상태'] == '넣음' for r in report)}개",
+        "",
+        "| 회사 | 만든 사람 | 프로젝트 | ID | 넣은 양식 | 상태 |",
+        "| --- | --- | --- | --- | --- | --- |",
+    ]
+    for row in report:
+        added = " · ".join(row["작업본"]) or "—"
+        lines.append(
+            f"| {row['company_id']} | {row.get('owner') or row['user_id']} | {row['name']} "
+            f"| `{row['id']}` | {added} | {row['상태']} |"
+        )
+    return "\n".join(lines) + "\n"
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--report", type=Path)
+    args = parser.parse_args()
+    await init_db_pool()
+    try:
+        projects = await _projects()
+    finally:
+        await close_db_pool()
+    report = migrate(projects, dry_run=args.dry_run)
+    text = render(report, layers.list_templates(layers.system_dir()))
+    if args.report:
+        args.report.write_text(text, encoding="utf-8")
+    sys.stdout.reconfigure(encoding="utf-8")
+    print(text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

resources/tester/test_m02_table_fill.py

@@ -0,0 +1,150 @@
+"""M02 구조물 집계표 채우기 — 펼침 열 · 측점 줄 · 관 줄 나눔 · 손 값 지킴 (임시 프로젝트)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from config import config_system
+from M02_MasterTemplete import M02_Table_Fill as fill
+
+MASTER = config_system.PROJECT_ROOT / "resources/master_template/table/구조물집계표.json"
+
+
+def _master() -> dict[str, Any]:
+    return json.loads(MASTER.read_text(encoding="utf-8"))
+
+
+@pytest.fixture
+def project(tmp_path: Path) -> Path:
+    root = tmp_path / "proj"
+    route = root / "B05_Profile/route"
+    route.mkdir(parents=True)
+    structures = [
+        {
+            "type_id": "masonry_dry",
+            "chainage_m": 85,
+            "start_m": 80,
+            "end_m": 90,
+            "options": {"height_m": 2.0, "length_m": 10},
+        },
+        {
+            "type_id": "masonry_wet",
+            "chainage_m": 80,
+            "start_m": 80,
+            "end_m": 90,
+            "options": {"height_m": 2.5, "length_m": 10},
+        },
+        {
+            "type_id": "pavement_concrete",
+            "chainage_m": 205,
+            "start_m": 200,
+            "end_m": 230,
+            "options": {"length_m": 30, "width_m": 3, "thickness_cm": 20, "joint_spacing_m": 5},
+        },
+        {"type_id": "position_sign", "chainage_m": 300},
+        {
+            "type_id": "boulder_masonry",
+            "chainage_m": 325,
+            "start_m": 320,
+            "end_m": 330,
+            "options": {"height_m": 2.5, "length_m": 10, "bond": "메쌓기", "stone_cm": "40~60"},
+        },
+    ]
+    for index, item in enumerate(structures):
+        item.setdefault("structure_id", f"st{index}")
+        item.setdefault("placement", "interval" if "start_m" in item else "point")
+    (route / "structures.json").write_text(
+        json.dumps({"revision": 1, "structures": structures}, ensure_ascii=False), encoding="utf-8"
+    )
+    edits = root / "B04_PreProcess/drainage/edits"
+    edits.mkdir(parents=True)
+    points = [
+        {"chainage_m": 85.05, "source": "user", "options": {"pipe_diameter_mm": "1000"}},
+        {"chainage_m": 85.05, "source": "user", "options": {"pipe_kind": "흄관"}},
+        {"chainage_m": 173.09, "source": "user", "facility": "ford_bridge", "options": {}},
+    ]
+    (edits / "pipe_points.json").write_text(json.dumps({"points": points}), encoding="utf-8")
+    return root
+
+
+def _filled(project: Path, document: dict[str, Any] | None = None) -> dict[str, Any]:
+    return fill.fill_table(project, document or _master(), {85.0: 14.0})
+
+
+def _col(doc: dict[str, Any], col_id: str) -> dict[str, Any]:
+    return next(column for column in doc["열"] if column["id"] == col_id)
+
+
+def test_펼침_열은_설계에_쓰인_값만_차례대로(project: Path) -> None:
+    doc = _filled(project)
+    ids = [column["id"] for column in doc["열"]]
+    assert [i for i in ids if i.startswith("ms|")] == ["ms|찰쌓기|2.5", "ms|메쌓기|2.0"]
+    assert _col(doc, "ms|메쌓기|2.0")["머리"] == ["돌쌓기", "메쌓기", "H=2.0"]
+    assert [i for i in ids if i.startswith("pp_len|")] == [
+        "pp_len|흄관|1000",
+        "pp_len|파형강관|1000",
+    ]
+    assert not any(i.startswith("rw|") for i in ids)  # 옹벽 없음 → 열 없음
+    assert "rv" not in ids and "ps" in ids  # 펼침 마스터 칸은 빠지고 고정 열은 남음
+    # 관 유입·유출 기슭막이는 레지스트리 기본값(찰 · 메 · 높이 없음)으로 열이 섬 · id 는 설계 값
+    assert "rv|돌쌓기(찰)|-" in ids and "rv|돌쌓기(메)|-" in ids
+    assert _col(doc, "rv|돌쌓기(찰)|-")["머리"] == ["돌기슭막이", "찰쌓기", "H=-"]
+    assert not any(i.startswith("pg_in|") for i in ids)  # 날개벽·집수정 형식 없음 → 관보호공 없음
+
+
+def test_계산_열_식은_같은_묶음_열로_바뀌고_빈_변수면_안_셈(project: Path) -> None:
+    doc = _filled(project)
+    assert _col(doc, "pv_a|20")["식"] == 'IF([pv_l|20]>0,[pv_b|20]*[pv_l|20]+[pv_w|20],"")'
+    cp = _col(doc, "pp_cp|파형강관|1000")["식"]
+    assert "[pp_len|파형강관|1000]" in cp and "[$파형강관_1본_m]" in cp
+    assert "식" not in _col(doc, "pp_cp|흄관|1000")  # 흄관 1본 길이 빈칸 → 계산 안 함
+    assert _col(doc, "bm_area|메쌓기|40~60|2.5")["식"] == (
+        'IF([bm_len|메쌓기|40~60|2.5]>0,[bm_len|메쌓기|40~60|2.5]*2.5,"")'
+    )
+
+
+def test_측점_줄_같은_측점은_한_줄_관은_나눔(project: Path) -> None:
+    doc = _filled(project)
+    rows = {row["id"]: row for row in doc["줄"]}
+    assert doc["줄"][0]["고정"] == "전구간"
+    interval = rows["s80.00~90.00"]["값"]
+    assert interval["sta"] == "NO.4~NO.4+10"
+    assert interval["ms|찰쌓기|2.5"] == 10 and interval["ms|메쌓기|2.0"] == 10
+    first, second = rows["s85.05"]["값"], rows["s85.05#1"]["값"]
+    assert first["sta"] == second["sta"] == "NO.4+5.05"
+    assert first["pp_len|파형강관|1000"] == 14.0 and "pp_len|흄관|1000" not in first
+    assert second["pp_len|흄관|1000"] == 14.0  # 한 측점 횡단 관 연장은 하나 — 두 관이 같이 씀
+    assert first["rv|돌쌓기(찰)|-"] == second["rv|돌쌓기(찰)|-"] == 10
+    assert rows["s300.00"]["값"]["ps"] == 1
+    assert [row["값"].get("no") for row in doc["줄"][1:]] == [str(i) for i in range(1, 8)]
+    assert rows["s200.00~230.00"]["식"]["pv_jt|20"].count("/5)") == 1  # 줄눈 간격 5 박음
+    assert doc["알림"] == []
+    bare = fill.fill_table(project, _master(), {})
+    assert bare["알림"] == ["관 연장 없음 2곳 — B06 횡단 설계 전이면 빈칸"]
+    assert not any("pp_len" in key for row in bare["줄"] for key in row["값"])
+
+
+def test_손_값과_전구간_줄은_다시_채워도_지킴(project: Path) -> None:
+    doc = _filled(project)
+    doc["줄"][0]["값"]["ps"] = 3
+    next(row for row in doc["줄"] if row["id"] == "s300.00")["값"]["h_intake"] = 2
+    doc["줄"].append({"id": "u1", "값": {"sta": "NO.9", "h_rockfall": 4}, "손": True})
+    doc["줄"].append({"id": "s999.00", "값": {"sta": "NO.49+19", "h_waste": 1.5}})
+    again = _filled(project, doc)
+    rows = {row["id"]: row for row in again["줄"]}
+    assert rows["all"]["값"]["ps"] == 3
+    assert rows["s300.00"]["값"]["h_intake"] == 2
+    assert rows["u1"]["값"]["h_rockfall"] == 4
+    assert rows["s999.00"]["손"] is True and rows["s999.00"]["값"]["h_waste"] == 1.5
+    assert [c["id"] for c in again["열"]] == [c["id"] for c in doc["열"]]  # 두 번 펼치지 않음
+
+
+def test_측점_표기() -> None:
+    assert fill.station_label(0) == "NO.0"
+    assert fill.station_label(38) == "NO.1+18"
+    assert fill.station_label(59.999) == "NO.3"
+    assert fill.station_label(2413.5) == "NO.120+13.5"

resources/tester/test_m02_template_layers.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 from typing import Any
 
@@ -181,3 +182,86 @@ def test_시스템_층은_읽기만_이름은_막음(world: dict[str, Any]) -> N
     for name in ("..", "_initial", "a.b/c"):
         with pytest.raises(ValueError):
             layers.template_path(layers.system_dir(), "table", name)
+
+
+def test_채운_표는_작업본을_채워_돌려주고_저장_안_함(
+    world: dict[str, Any], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    async def designs(project_id: str) -> list[dict[str, Any]]:
+        return [{"chainage_m": 40.0, "design": {"pipe_length_m": 12}}]
+
+    monkeypatch.setattr(router_module, "_cross_designs", designs)
+    monkeypatch.setattr(router_module.fill, "recalc", lambda document: {"계산": {}})
+    root = _root(world, P1)
+    master = config_system.PROJECT_ROOT / "resources/master_template/table/구조물집계표.json"
+    layers.write_template(
+        root / "templates", "table", "구조물집계표", json.loads(master.read_text(encoding="utf-8"))
+    )
+    edits = root / "B04_PreProcess/drainage/edits"
+    edits.mkdir(parents=True, exist_ok=True)
+    (edits / "pipe_points.json").write_text(
+        json.dumps({"points": [{"chainage_m": 40.0, "source": "user"}]}), encoding="utf-8"
+    )
+    before = layers.version_of(root / "templates/table/구조물집계표.json")
+    got = world["client"].get(f"/api/m02/projects/{P1}/tables/구조물집계표/filled").json()
+    row = next(r for r in got["문서"]["줄"] if r["id"] == "s40.00")
+    assert row["값"]["sta"] == "NO.2" and row["값"]["pp_len|파형강관|1000"] == 12
+    assert got["결과"] == {"계산": {}} and got["판"] == before
+    assert layers.version_of(root / "templates/table/구조물집계표.json") == before
+
+
+def test_프로젝트_만들기_자리에서_시스템_양식을_복사(world: dict[str, Any], tmp_path: Path) -> None:
+    from B02_ProjRegister.B02_ProjRegister_Repository import _initialize_project_storage
+
+    layers.write_template(layers.company_dir(7), "table", "구조물집계표", {"회사": 1})
+    root = tmp_path / "new_project"
+    _initialize_project_storage(root, "new")
+    got = layers.read_template(root / "templates", "table", "구조물집계표")
+    assert got["문서"] == {"판": 1, "열": []}  # 회사 공식이 있어도 시스템
+    assert (root / "templates/_initial/drawing/A1_도각.json").is_file()
+    assert (root / "project_manifest.json").is_file()
+
+
+def test_옛_프로젝트_넣기_도구는_더하기만_폴더_없으면_건너뜀(world: dict[str, Any]) -> None:
+    from M02_MasterTemplete import M02_Template_Migrate as migrate
+
+    old = world["storage"] / "7/42/old-project"
+    (old / "B05_Profile").mkdir(parents=True)
+    (old / "B05_Profile/keep.txt").write_text("x", encoding="utf-8")
+    root1 = _root(world, P1)
+    layers.write_template(root1 / "templates", "table", "구조물집계표", {"고침": 1})
+    (root1 / "templates/drawing/A1_도각.json").unlink()
+    projects = [
+        {
+            "id": "old",
+            "name": "옛",
+            "company_id": 7,
+            "user_id": 42,
+            "storage_path": "storage/7/42/old-project",
+        },
+        {
+            "id": P1,
+            "name": "첫째",
+            "company_id": 7,
+            "user_id": 42,
+            "storage_path": f"storage/7/42/{P1}",
+        },
+        {
+            "id": "gone",
+            "name": "없음",
+            "company_id": 7,
+            "user_id": 42,
+            "storage_path": "storage/7/42/gone",
+        },
+    ]
+    dry = migrate.migrate(projects, dry_run=True)
+    assert not (old / "templates").exists() and dry[0]["상태"] == "넣을 것(시험)"
+    report = {row["id"]: row for row in migrate.migrate(projects)}
+    assert report["old"]["상태"] == "넣음" and (old / "templates/_initial/table").is_dir()
+    assert (old / "B05_Profile/keep.txt").read_text(encoding="utf-8") == "x"
+    assert report[P1]["작업본"] == ["drawing/A1_도각"]
+    assert layers.read_template(root1 / "templates", "table", "구조물집계표")["문서"] == {"고침": 1}
+    assert report["gone"]["상태"].startswith("폴더 없음")
+    assert not (world["storage"] / "7/42/gone").exists()
+    assert migrate.migrate(projects)[0]["상태"] == "이미 있음"
+    assert "| 7 | 42 | 옛 |" in migrate.render(list(report.values()), [])