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

B01_Dashboard/B01_Dashboard_Router_Home.py

@@ -140,7 +140,8 @@ async def _company(cursor: Any, company_id: int) -> dict[str, Any]:
 
 async def _forest(cursor: Any, company_id: int) -> dict[str, Any]:
     """예정노선 수 — 분석 전 = 노선을 아직 안 그은 프로젝트 · 분석됨 · 확정 = 연도(설계 내용 파일) ·
-    업체 지정 대기 = 확정 연도 중 그 연도 조각에 업체가 하나도 없는 것."""
+    업체 지정 대기 = 확정 연도 중 그 연도 세트에 업체가 하나도 없는 것(업체 줄 = 세트 열쇠 ·
+    옛 조각 id = 든 세트)."""
     from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
 
     await cursor.execute(
@@ -161,11 +162,13 @@ async def _forest(cursor: Any, company_id: int) -> dict[str, Any]:
             continue
         if not state["plan"]["lines"]:
             status["draft"] += 1
+        found = (store.find_unit(state["units"], key) for key in pieces)
+        held = {unit["id"] for unit in found if unit}
         for year, row in state["years"].items():
             key = "confirmed" if row["confirmed_at"] else "analyzed"
             status[key] += 1
-            ids = {p["id"] for p in row["route"]["pieces"]}
-            waiting += bool(row["confirmed_at"] and not ids & pieces)
+            ids = {u["id"] for u in state["units"] if u["year"] == year}
+            waiting += bool(row["confirmed_at"] and not ids & held)
     return {"plan_status": status, "assign_waiting": waiting}
 
 

D01_PlanRegister/D01_PlanRegister_Router.py

@@ -78,11 +78,11 @@ async def post_plan_project(
 async def assigned_plan_routes(
     session: dict[str, Any] = Depends(require_company),
 ) -> dict[str, Any]:
-    """우리 회사에 지정된 예정노선 조각 — 확정된 것(또는 이미 연결한 것) · 연결 상태 · 업데이트 ·
-    조각 이름 · 길이 · 연도(설계 내용 파일 `plan_state` · 업체 줄은 조각 번호만)."""
+    """우리 회사에 지정된 예정노선 세트 — 확정된 것(또는 이미 연결한 것) · 연결 상태 · 업데이트 ·
+    세트 이름 「2027-1」 · 길이 · 연도(설계 내용 파일 `plan_state` · 업체 줄은 세트 열쇠만 ·
+    옛 조각 id = 든 세트 · PLAN 638-1)."""
     from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
     from D04_PlanConfirm.D04_PlanConfirm_Link import update_available
-    from D04_PlanConfirm.D04_PlanConfirm_Router import piece_label
 
     async with (
         get_db_pool().acquire() as connection,
@@ -100,18 +100,17 @@ async def assigned_plan_routes(
             (session.get("company_id"),),
         )
         rows = await cursor.fetchall()
-    pieces: dict[tuple[str, str], dict[str, Any]] = {}
+    sets: dict[str, list[dict[str, Any]]] = {}
     for project_id in {str(r["project_id"]) for r in rows}:
         try:
-            units = (await store.plan_state(project_id))["units"]
+            sets[project_id] = (await store.plan_state(project_id))["units"]
         except HTTPException:  # 노선 파일이 틀린 프로젝트 = 건너뜀(목록 전체를 막지 않음)
             continue
-        for unit in units:
-            pieces[(project_id, unit["id"])] = unit
     routes = []
     for row in rows:
         fingerprint = row.pop("source_fingerprint")
-        piece = pieces.get((str(row["project_id"]), row["piece_id"]))
+        piece = store.find_unit(sets.get(str(row["project_id"]), []), row["piece_id"])
+        row["piece_id"] = piece["id"] if piece else row["piece_id"]
         confirmed_at = piece and piece["confirmed_at"]
         if confirmed_at is None and row["linked_project_id"] is None:
             continue
@@ -121,8 +120,7 @@ async def assigned_plan_routes(
                 **row,
                 "year": piece and piece["year"],
                 "confirmed_at": confirmed_at,
-                "piece_label": piece
-                and piece_label(piece["line_name"], piece["from_m"], piece["to_m"]),
+                "piece_label": piece and piece["id"],
                 "length_m": piece and piece["length_m"],
                 "update_available": update_available(route, fingerprint, bool(confirmed_at)),
             }

D03_PlanAnalysis/D03_PlanAnalysis_Network.py

@@ -7,12 +7,18 @@
   다시 셈 → 파일(원자 쓰기) · 연도 상태 줄(`plan_route_years` 지문 · 확정) · 사업연도 끝
 
 길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
-조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
+조각 = 선의 마크 사이 구간(구역 × 갈래) · 조각마다 세트 번호 `set`(요구 547 · 548 · PLAN 638-1) —
+세트 열쇠 「연도-번호」(2027-1) = 같은 노선 · 업체 지정 단위(`plan_route_assignees.piece_id`) ·
+같은 선 이웃 조각이 같은 세트면 [저장] · 읽기 때 하나로(앞 조각 id · 사이 마크 걷음) ·
+`set` 없는 옛 조각 = 그 해 선 차례 · 거리 차례로 최댓값 + 1 ·
+세트마다 덩이 수 · 갈래 = `derived.sets`.
 지점(`points` · 긋기에 붙는 점) = 같은 파일 · [저장] 몸에 실림(빈 값 = 그대로) ·
 기준점(`projects.site_*`) = 처음 자료 받는 기준일 뿐(노선 · 붙음 무관 · PLAN 591-2) ·
 [저장] 몸 `site` 가 바뀌었으면 같은 칸을 고침(594-3 · 등록과 한 칸).
 지점 · 선 이름(20 자 · 파일 이름 못 쓰는 글자 막음) · 선 노드 기록 = 파일에 있을 때만(594-1).
 계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
+⚠ 짝 = `normalize`(세트 번호 · 합침) ↔ TS `logic/D03_PlanAnalysis_Group.ts` —
+거울 시험 `test_638_sets`(사례 `resources/tester/fixtures/d03_638_set_cases.json`).
 """
 
 from __future__ import annotations
@@ -85,6 +91,7 @@ class Piece(BaseModel):
     from_m: float
     to_m: float
     year: int
+    set: int | None = Field(default=None, ge=1)  # 세트 번호(638-1) · 빈 값 = 그 해 새 번호
 
 
 class Network(BaseModel):
@@ -168,11 +175,137 @@ def _check(ok: bool, message: str) -> None:
         raise NetworkError(message)
 
 
+def set_key(year: int, no: int) -> str:
+    return f"{int(year)}-{int(no)}"
+
+
+def _numbered(network: dict[str, Any]) -> list[dict[str, Any]]:
+    """조각 사본 — (선 차례, from_m) 차례 · `set` 없는 조각 = 그 해 최댓값 + 1(차례대로)."""
+    order = {line["id"]: i for i, line in enumerate(network["lines"])}
+    pieces = sorted(
+        (dict(p) for p in network["pieces"]),
+        key=lambda p: (order.get(p["line"], len(order)), p["from_m"]),
+    )
+    top: dict[int, int] = {}
+    for piece in pieces:
+        if piece.get("set"):
+            top[piece["year"]] = max(top.get(piece["year"], 0), int(piece["set"]))
+    for piece in pieces:
+        if not piece.get("set"):
+            top[piece["year"]] = piece["set"] = top.get(piece["year"], 0) + 1
+    return pieces
+
+
+def normalize(network: dict[str, Any]) -> dict[str, Any]:
+    """세트 번호 매김 + 같은 선 이웃 같은 세트 합침(앞 조각 id · 사이 마크 걷음) — 사본.
+
+    ⚠ 짝 = TS `logic/D03_PlanAnalysis_Group.ts` — 거울 시험 `test_638_sets`."""
+    marks = [dict(m) for m in network["marks"]]
+    pieces: list[dict[str, Any]] = []
+    for piece in _numbered(network):
+        prev = pieces[-1] if pieces else None
+        same = (
+            prev is not None
+            and prev["line"] == piece["line"]
+            and (prev["year"], prev["set"]) == (piece["year"], piece["set"])
+            and abs(prev["to_m"] - piece["from_m"]) <= PIECE_TOL_M
+        )
+        cut = same and next(
+            (
+                m
+                for m in marks
+                if m["line"] == piece["line"] and abs(m["at_m"] - piece["from_m"]) <= PIECE_TOL_M
+            ),
+            None,
+        )
+        if cut:
+            marks.remove(cut)
+            prev["to_m"] = piece["to_m"]
+            continue
+        pieces.append(piece)
+    return {**network, "marks": marks, "pieces": pieces}
+
+
+def _touch(a: Any, b: Any) -> bool:
+    return draw.segment_length(a, b) <= PIECE_TOL_M
+
+
+def _set_shape(items: list[dict[str, Any]]) -> tuple[int, bool]:
+    """(덩이 수, 갈래 있음) — 조각 끝이 다른 조각 꼭짓점에 닿으면 이어짐 · 한 점에 3 이상 = 갈래."""
+    root = list(range(len(items)))
+
+    def find(i: int) -> int:
+        while root[i] != i:
+            i = root[i]
+        return i
+
+    branch = False
+    for i, piece in enumerate(items):
+        for end in (piece["planned"][0], piece["planned"][-1]):
+            degree = 0
+            for j, other in enumerate(items):
+                line = other["planned"]
+                hits = sum(_touch(end, p) for p in (line[0], line[-1]))
+                hits += 2 * any(_touch(end, p) for p in line[1:-1])
+                degree += hits
+                if hits and j != i:
+                    root[find(i)] = find(j)
+            branch = branch or degree >= 3
+    return len({find(i) for i in range(len(items))}), branch
+
+
+def join_set(items: list[dict[str, Any]]) -> list[list[float]] | None:
+    """세트 조각을 끝끼리 이은 한 꺾은선(첫 조각 방향) · 못 이으면 None."""
+    chain = [list(p) for p in items[0]["planned"]]
+    rest = list(items[1:])
+    while rest:
+        for piece in rest:
+            line = [list(p) for p in piece["planned"]]
+            if _touch(chain[-1], line[0]):
+                chain += line[1:]
+            elif _touch(chain[-1], line[-1]):
+                chain += line[::-1][1:]
+            elif _touch(chain[0], line[-1]):
+                chain = line[:-1] + chain
+            elif _touch(chain[0], line[0]):
+                chain = line[::-1][:-1] + chain
+            else:
+                continue
+            rest.remove(piece)
+            break
+        else:
+            return None
+    return chain
+
+
+def sets_of(pieces: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """세트마다 {key, year, no, parts, branch, length_m, pieces} — 연도 · 번호 차례."""
+    groups: dict[tuple[int, int], list[dict[str, Any]]] = {}
+    for piece in pieces:
+        groups.setdefault((piece["year"], piece["set"]), []).append(piece)
+    out = []
+    for (year, no), items in sorted(groups.items()):
+        parts, branch = _set_shape(items)
+        out.append(
+            {
+                "key": set_key(year, no),
+                "year": year,
+                "no": no,
+                "parts": parts,
+                "branch": branch,
+                "length_m": sum(p["length_m"] for p in items),
+                "pieces": [p["id"] for p in items],
+            }
+        )
+    return out
+
+
 def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, Any]:
-    """선 · 마크 · 조각 → 선 길이 · 측점 · 조각(경계 = 마크 · 꺾은선 · 측점) ·
-    틀리면 `NetworkError`.
+    """선 · 마크 · 조각 → 선 길이 · 측점 · 조각(경계 = 마크 · 꺾은선 · 측점) · 세트 ·
+    틀리면 `NetworkError` · 세트 번호 · 합침은 `normalize` 뒤 값으로.
 
     `year_from` = 사업연도 시작(그 앞 연도 조각 거절 · None = 안 봄)."""
+    network = normalize(network)
     lines: dict[str, list] = {}
     for line in network["lines"]:
         vertices = line["vertices"]
@@ -239,6 +372,8 @@ def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, A
                 "id": piece["id"],
                 "line": key,
                 "year": int(piece["year"]),
+                "set": int(piece["set"]),
+                "set_key": set_key(piece["year"], piece["set"]),
                 "from_m": a,
                 "to_m": b,
                 "length_m": b - a,
@@ -255,6 +390,7 @@ def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, A
             {"id": key, "length_m": lengths[key], "stations": stations[key]} for key in lines
         ],
         "pieces": pieces,
+        "sets": sets_of(pieces),
     }
 
 
@@ -326,7 +462,7 @@ def base_view(plan: dict[str, Any], site: dict[str, Any] | None) -> dict[str, An
 
 
 def _view(plan: dict[str, Any], project: dict[str, Any]) -> dict[str, Any]:
-    network = {key: plan[key] for key in ("epsg", "lines", "marks", "pieces")}
+    network = normalize({key: plan[key] for key in ("epsg", "lines", "marks", "pieces")})
     return {
         "status": "success",
         "base": base_view(plan, project["site"]),
@@ -364,21 +500,42 @@ async def get_plan_network(project_id: UUID) -> dict[str, Any]:
         raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
 
 
+def old_keys(plan: dict[str, Any]) -> dict[str, str]:
+    """저장 전 파일의 조각 id → 세트 열쇠(옛 업체 줄 `p1` … 풀어 봄 · 합쳐질 뒤 조각 포함)."""
+    return {p["id"]: set_key(p["year"], p["set"]) for p in _numbered(plan)}
+
+
 async def _store_status(
-    connection: Any, project_id: UUID, pieces: list[dict[str, Any]], years: dict[int, Any]
+    connection: Any,
+    project_id: UUID,
+    keys: set[str],
+    years: dict[int, Any],
+    before: dict[str, str] | None = None,
 ) -> None:
-    """연도 상태 줄 — 있는 연도만(지문이 바뀐 연도만 확정 풂) · 없는 연도 줄 지움 · 사업연도 끝."""
+    """업체 줄 열쇠 · 연도 상태 줄 — 지정된 세트가 사라지면 409 · 옛 조각 열쇠 = 그 세트 열쇠로 ·
+    있는 연도만(지문이 바뀐 연도만 확정 풂) · 없는 연도 줄 지움 · 사업연도 끝."""
     pid = str(project_id)
+    before = before or {}
     async with connection.cursor() as cursor:
         await cursor.execute(
             "SELECT DISTINCT piece_id FROM plan_route_assignees"
             " WHERE project_id = %s AND piece_id IS NOT NULL",
             (pid,),
         )
-        lost = sorted({r[0] for r in await cursor.fetchall()} - {p["id"] for p in pieces})
+        held = {r[0] for r in await cursor.fetchall()}
+        moved = {k: before[k] for k in held if k not in keys and k in before}
+        lost = sorted({moved.get(k, k) for k in held} - keys)
         if lost:
             raise HTTPException(
-                status_code=409, detail=f"업체가 지정된 조각을 지울 수 없습니다: {lost}"
+                status_code=409,
+                detail=f"업체가 지정된 세트가 없어집니다: {', '.join(lost)}"
+                " — 결과 확인(D04)에서 업체를 먼저 빼세요.",
+            )
+        for old, new in moved.items():
+            await cursor.execute(
+                "UPDATE plan_route_assignees SET piece_id = %s"
+                " WHERE project_id = %s AND piece_id = %s",
+                (new, pid, old),
             )
         for year, row in years.items():
             # confirmed_at 을 fingerprint 보다 먼저 셈
@@ -447,6 +604,8 @@ async def put_plan_network(
         if plan["epsg"] is not None and plan["epsg"] != network["epsg"]:
             raise HTTPException(status_code=400, detail="좌표계가 저장된 노선과 다릅니다.")
         project = await project_info(project_id)
+        before = old_keys(plan)
+        network = normalize(network)
         try:
             derived = derive(network, project["year_start"])
         except NetworkError as error:
@@ -460,7 +619,11 @@ async def put_plan_network(
             await connection.begin()
             try:
                 await _store_status(
-                    connection, project_id, derived["pieces"], year_rows(derived["pieces"])
+                    connection,
+                    project_id,
+                    {s["key"] for s in derived["sets"]},
+                    year_rows(derived["pieces"]),
+                    before,
                 )
                 if site:
                     await _store_site(connection, project_id, site)

D03_PlanAnalysis/D03_PlanAnalysis_Store.py

@@ -1,11 +1,13 @@
 """D 예정노선 설계 내용 한 파일 — `<프로젝트>/D03_PlanAnalysis/plan_network.json`
 (요구 452 · 453 · PLAN 590-1).
 
-- 꼴 = `{version, epsg, points: [{x, y}], lines, marks, pieces: [{id, line, from_m, to_m, year}]}` —
-  좌표계 = 머리 한 곳 · 연도 = 조각의 `year` 뿐(빈 연도 줄 없음) · 길이 · 측점 · 조각 좌표 ·
-  연도 지문은 읽을 때 `plan_state` 가 셈(저장 안 함).
+- 꼴 = `{version: 2, epsg, points: [{x, y}], lines, marks,
+  pieces: [{id, line, from_m, to_m, year, set}]}` — 좌표계 = 머리 한 곳 · 연도 = 조각의 `year` 뿐 ·
+  세트 = 조각의 `set`(열쇠 「연도-번호」 · PLAN 638-1 · 판 1 = `set` 없음 → 읽을 때 번호) ·
+  길이 · 측점 · 조각 좌표 · 세트 덩이 · 연도 지문은 읽을 때 `plan_state` 가 셈(저장 안 함).
 - DB 에 남는 것 = 등록 칸 · 연도 상태(`plan_route_years` 의 year · fingerprint · confirmed_at) ·
-  업체 줄(`plan_route_assignees` 조각 번호 · 업체 · 메일 시각 · 연결 — 연도는 이 파일에서).
+  업체 줄(`plan_route_assignees.piece_id` = 세트 열쇠(옛 줄 = 조각 id → `find_unit` 이 그 세트로) ·
+  업체 · 메일 시각 · 연결 — 연도는 이 파일에서).
 - 파일이 없으면 빈 설계(새 프로젝트) · 파일은 첫 쓰기 때 생김(옛 DB 칸 옮김 = 590-3 에서 끝남).
 - 쓰기 = `project_lock(project_id)` 안에서 `save_plan`(같은 폴더 임시 파일 → 바꿔치기 · 반쯤 쓴
   파일이 안 보임) · 읽기는 잠금 없음.
@@ -30,7 +32,7 @@ from D03_PlanAnalysis import D03_PlanAnalysis_Network as network
 from D04_PlanConfirm.D04_PlanConfirm_Repository import line_names
 
 FILE = Path("D03_PlanAnalysis") / "plan_network.json"
-VERSION = 1
+VERSION = 2  # 2 = 조각 `set`(옛 서버가 덮어 세트를 지우지 않게 · 638-1)
 KEYS = ("epsg", "points", "lines", "marks", "pieces")
 
 
@@ -101,16 +103,59 @@ async def project_row(project_id: UUID | str) -> dict[str, Any]:
         return await _project(cursor, project_id, plan_only=False)
 
 
-def plan_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
-    """조각마다 {id · line · line_name · year · epsg · from_m · to_m · length_m · planned ·
-    stations}
-    — 연도 · 선 차례 · 선 위 거리 차례 · 틀리면 `NetworkError`."""
-    pieces = network.derive(plan)["pieces"] if plan["lines"] else []
+def plan_parts(plan: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """(조각, 세트 단위) — 틀리면 `NetworkError`.
+
+    조각 = {id · line · line_name · year · set · set_key · epsg · from_m · to_m · length_m ·
+    planned · stations} · 연도 · 선 차례 · 선 위 거리 차례.
+    세트 단위 = {id(「2027-1」) · set · year · line · line_name(첫 조각) · epsg · from_m · to_m ·
+    length_m · planned(이은 한 줄 · 덩이 2 이상 · 갈래면 빈 것) · parts · branch ·
+    pieces(조각 id)} —
+    여러 조각 세트의 to_m = from_m + 길이."""
+    if not plan["lines"]:
+        return [], []
+    derived = network.derive(plan)
+    pieces = derived["pieces"]
     names = dict(zip((line["id"] for line in plan["lines"]), line_names(plan["lines"])))
     for piece in pieces:
         piece.update(line_name=names.get(piece["line"], "갈래"), epsg=plan["epsg"])
     pieces.sort(key=lambda p: p["year"])  # 같은 연도 안 = derive 차례(선 · 거리) 그대로
-    return pieces
+    by_id = {p["id"]: p for p in pieces}
+    units = []
+    for item in derived["sets"]:
+        parts = [by_id[i] for i in item["pieces"]]
+        first = parts[0]
+        whole = item["parts"] == 1 and not item["branch"]
+        units.append(
+            {
+                "id": item["key"],
+                "set": item["no"],
+                "year": item["year"],
+                "line": first["line"],
+                "line_name": first["line_name"],
+                "epsg": plan["epsg"],
+                "from_m": first["from_m"],
+                "to_m": first["to_m"] if len(parts) == 1 else first["from_m"] + item["length_m"],
+                "length_m": item["length_m"],
+                "planned": (network.join_set(parts) or []) if whole else [],
+                "parts": item["parts"],
+                "branch": item["branch"],
+                "pieces": item["pieces"],
+            }
+        )
+    return pieces, units
+
+
+def plan_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
+    """조각만(`plan_parts` 앞 값)."""
+    return plan_parts(plan)[0]
+
+
+def find_unit(units: list[dict[str, Any]], key: str | None, field: str = "id") -> Any:
+    """업체 줄 열쇠 → 세트 단위 — 세트 열쇠 · 옛 조각 id(그 조각이 든 세트) · 없으면 None."""
+    return next((u for u in units if u[field] == key), None) or next(
+        (u for u in units if key in u["pieces"]), None
+    )
 
 
 async def year_status(project_id: UUID | str) -> dict[int, dict[str, Any]]:
@@ -127,19 +172,20 @@ async def year_status(project_id: UUID | str) -> dict[int, dict[str, Any]]:
 
 
 async def plan_state(project_id: UUID | str) -> dict[str, Any]:
-    """{root, plan, units, years} — `years` = `year_rows` + `confirmed_at`(지문이 같을 때만) ·
-    단위마다 `confirmed_at` · 노선이 틀리면 409."""
+    """{root, plan, pieces, units, years} — `units` = 세트 단위(`plan_parts`) ·
+    `years` = `year_rows` + `confirmed_at`(지문이 같을 때만) · 단위 · 조각마다 `confirmed_at` ·
+    노선이 틀리면 409."""
     root, plan = await load_plan(project_id)
     try:
-        units = plan_units(plan)
+        pieces, units = plan_parts(plan)
     except network.NetworkError as error:
         raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
-    years = network.year_rows(units)
+    years = network.year_rows(pieces)
     status = await year_status(project_id)
     for year, row in years.items():
         saved = status.get(year)
         same = saved is not None and saved["fingerprint"] == row["fingerprint"]
         row["confirmed_at"] = saved["confirmed_at"] if same else None
-    for unit in units:
+    for unit in (*pieces, *units):
         unit["confirmed_at"] = years[unit["year"]]["confirmed_at"]
-    return {"root": root, "plan": plan, "units": units, "years": years}
+    return {"root": root, "plan": plan, "pieces": pieces, "units": units, "years": years}

D04_PlanConfirm/D04_PlanConfirm_Link.py

@@ -4,12 +4,13 @@
   new = 임도 프로젝트 새로(B02 저장소 함수) · existing = 고른 우리 회사 임도 프로젝트
 - `GET  /api/projects/{id}/plan-source` — 임도 프로젝트가 이어 받은 예정노선 · 업데이트 있음
 
-연결 = 지정 행의 조각 shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에
+연결 = 지정 행의 세트(「2027-1」 · PLAN 638-1) 구간을 이은 한 줄 shp 한 벌(D04 내보내기 함수)을
+B03 입력 `input/shp/` 에
 원래 이름으로 + `input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길.
 입력이 다 차면 B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림.
 연결 시점 복사 · 원본이 바뀌면 알림만(내용 지문 비교 · 자동 반영 없음).
-조각 노선 · 연도 · 확정 = 설계 내용 파일(`D03_PlanAnalysis_Store.plan_state`) —
-업체 줄은 조각 번호만.
+세트 노선 · 연도 · 확정 = 설계 내용 파일(`D03_PlanAnalysis_Store.plan_state`) —
+업체 줄은 세트 열쇠만(옛 조각 id = 그 조각이 든 세트) · 덩이 2 이상 · 갈래 세트는 연결 거절.
 """
 
 from __future__ import annotations
@@ -102,20 +103,26 @@ async def _fetch_one(sql: str, args: tuple) -> dict[str, Any] | None:
 
 
 async def piece_route(project_id: str, piece_id: str | None) -> dict[str, Any]:
-    """그 조각(파일) — {year, epsg, route: {planned, 누가 · 길이}, confirmed_at} ·
-    조각이 없으면 노선 · 연도 · 확정 None."""
+    """그 세트(파일 · 옛 조각 id = 든 세트) — {set_key, year, epsg, route: {planned(이은 한 줄),
+    누가 · 길이}, parts, branch, confirmed_at} · 세트가 없으면 노선 · 연도 · 확정 None."""
     state = await store.plan_state(project_id)
-    unit = next((u for u in state["units"] if u["id"] == piece_id), None)
+    unit = store.find_unit(state["units"], piece_id)
     if unit is None:
         return {
+            "set_key": None,
             "year": None,
             "epsg": state["plan"]["epsg"],
             "route": None,
+            "parts": 0,
+            "branch": False,
             "confirmed_at": None,
         }
     return {
+        "set_key": unit["id"],
         "year": unit["year"],
         "epsg": unit["epsg"],
+        "parts": unit["parts"],
+        "branch": unit["branch"],
         "route": {
             "planned": unit["planned"],
             "start_chainage_m": unit["from_m"],
@@ -147,7 +154,7 @@ async def load_road_project(project_id: str) -> dict[str, Any] | None:
 
 
 async def record_link(project_id: str, assignee: dict[str, Any], fingerprint: str | None) -> None:
-    """원본 칸 · 연결 표시 — 한 임도 프로젝트는 한 조각에만 이어짐(옛 연결은 끊음)."""
+    """원본 칸 · 연결 표시 — 한 임도 프로젝트는 한 세트에만 이어짐(옛 연결은 끊음)."""
     async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
         await connection.begin()
         try:
@@ -172,7 +179,7 @@ async def record_link(project_id: str, assignee: dict[str, Any], fingerprint: st
 
 
 def year_files(assignee: dict[str, Any]) -> dict[str, bytes]:
-    """그 조각 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
+    """그 세트 한 줄 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
     body = _parsed(assignee["route"]) or {}
     route = YearRoute(
         year=int(assignee["year"]),
@@ -296,12 +303,19 @@ async def _new_road_project(assignee: dict[str, Any], session: dict[str, Any]) -
 async def link_plan_route(
     assignee_id: int, payload: LinkRequest, session: dict[str, Any] = Depends(require_company)
 ) -> dict[str, Any] | JSONResponse:
-    """지정받은 조각 노선을 임도 프로젝트 B03 입력으로 — 지정 회사만."""
+    """지정받은 세트 노선(이은 한 줄)을 임도 프로젝트 B03 입력으로 — 지정 회사만."""
     assignee = await load_assignee(assignee_id)
     if not assignee:
         raise HTTPException(status_code=404, detail="지정 업체 행을 찾을 수 없습니다.")
     if assignee["company_id"] is None or assignee["company_id"] != session.get("company_id"):
         raise HTTPException(status_code=403, detail="이 예정노선을 지정받은 회사가 아닙니다.")
+    if assignee["parts"] > 1 or assignee["branch"]:
+        why = f"{assignee['parts']} 덩이로 끊김" if assignee["parts"] > 1 else "갈래 있음"
+        raise HTTPException(
+            status_code=409,
+            detail=f"세트 {assignee['set_key']} 가 한 줄이 아닙니다({why})"
+            " — 산림청이 예정노선에서 이어야 연결할 수 있습니다.",
+        )
     if not assignee["confirmed_at"] or route_fingerprint(assignee["route"]) is None:
         raise HTTPException(status_code=409, detail="확정된 노선이 없습니다.")
     if payload.mode == "existing":
@@ -333,7 +347,7 @@ async def link_plan_route(
         "status": "success",
         "project_id": project_id,
         "year": assignee["year"],
-        "piece_id": assignee.get("piece_id"),
+        "piece_id": assignee["set_key"] or assignee.get("piece_id"),
         **result,
     }
 
@@ -360,7 +374,7 @@ async def plan_source(project_id: UUID) -> dict[str, Any]:
         "assignee_id": row["assignee_id"],
         "plan_name": row["plan_name"],
         "year": row["year"],
-        "piece_id": row.get("piece_id"),
+        "piece_id": row["set_key"] or row.get("piece_id"),
         "update_available": update_available(
             row["route"], row["source_fingerprint"], row["confirmed_at"] is not None
         ),

D04_PlanConfirm/D04_PlanConfirm_Repository.py

@@ -84,9 +84,9 @@ async def apply_assignee_changes(
                 )
             for row in updates:
                 await cursor.execute(
-                    "UPDATE plan_route_assignees SET company_name = %s, email = %s"
-                    " WHERE id = %s AND project_id = %s",
-                    (row["company_name"], row["email"], row["id"], project_id),
+                    "UPDATE plan_route_assignees SET piece_id = %s, company_name = %s, email = %s"
+                    " WHERE id = %s AND project_id = %s",  # 옛 조각 id → 세트 열쇠(638-1)
+                    (row["piece_id"], row["company_name"], row["email"], row["id"], project_id),
                 )
             for row in inserts:
                 await cursor.execute(

D04_PlanConfirm/D04_PlanConfirm_Router.py

@@ -1,14 +1,16 @@
 """D 예정노선 확정 · 업체 지정 API (요구 345 · 391 · 452 · PLAN 531-1 · 531-2 · 563-2 · 590-1).
 
-- `GET  /api/projects/{id}/plan/assignees` — 조각별 지정 업체(지정 회사 = 자기 행만)
-- `PUT  /api/projects/{id}/plan/assignees` — 조각별 지정 업체 저장(등록 회사 또는 업체명 + 메일)
-- `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 조각 zip → 메일 첨부 → 발송 시각
-- `GET  /api/projects/{id}/plan/summary` — D04 화면 연도 · 조각 요약(지정 회사 = 자기 조각 · 행만)
-- `GET  /api/projects/{id}/plan/export` — [내보내기] zip(산림청 = 전 조각 + 전체 · 공유 = 자기 조각)
+- `GET  /api/projects/{id}/plan/assignees` — 세트별 지정 업체(지정 회사 = 자기 행만)
+- `PUT  /api/projects/{id}/plan/assignees` — 세트별 지정 업체 저장(등록 회사 또는 업체명 + 메일)
+- `POST /api/projects/{id}/plan/finalize` — 연도 노선 확정 → 업체별 세트 zip → 메일 첨부 → 발송 시각
+- `GET  /api/projects/{id}/plan/summary` — D04 화면 연도 · 세트 요약(지정 회사 = 자기 세트 · 행만)
+- `GET  /api/projects/{id}/plan/export` — [내보내기] zip(산림청 = 전 세트 + 전체 · 공유 = 자기 세트)
 
-단위 = 조각(설계 내용 파일 `D03_PlanAnalysis_Store` · 구역 × 갈래 · 요구 391). 확정은 연도 단위
-(`plan_route_years.confirmed_at` · 지문이 파일과 같을 때만). 업체 줄 = 조각 번호 ·
-연도는 파일 조각에서.
+단위 = 세트 「2027-1」(설계 내용 파일 `D03_PlanAnalysis_Store.plan_parts` · 요구 547 · 548 ·
+PLAN 638-1 · 칸 이름 `piece_id` = 세트 열쇠 그대로). 확정은 연도 단위
+(`plan_route_years.confirmed_at` · 지문이 파일과 같을 때만 · 덩이 2 이상 · 갈래 세트가 든 연도는
+막음). 업체 줄 = 세트 열쇠(옛 조각 id = 읽을 때 그 세트로 · 다음 [저장] 때 세트 열쇠로) ·
+연도는 파일에서.
 
 지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이
 GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 파일 노선을 서버가 다시 센 것으로만.
@@ -73,7 +75,7 @@ def _require_forest(session: dict[str, Any]) -> None:
 
 
 def _key(row: dict[str, Any]) -> tuple:
-    """같은 업체 판정 — 같은 조각에서 등록 회사 = id · 수기 = 메일(대소문자 무시)."""
+    """같은 업체 판정 — 같은 세트에서 등록 회사 = id · 수기 = 메일(대소문자 무시)."""
     if row.get("company_id") is not None:
         return (row["piece_id"], "c", row["company_id"])
     return (row["piece_id"], "e", (row.get("email") or "").strip().lower())
@@ -127,10 +129,14 @@ async def _plan_project(project_id: UUID) -> dict[str, Any]:
 
 
 def route_units(pieces: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """표 한 줄 단위 = 파일 조각(`store.plan_units` · 연도 차례)."""
+    """표 한 줄 단위 = 세트(`store.plan_parts` 세트 단위 · 연도 · 번호 차례)."""
     return [
         {
             "piece_id": p["id"],
+            "set": p.get("set"),
+            "parts": p.get("parts", 1),
+            "branch": p.get("branch", False),
+            "pieces": p.get("pieces", [p["id"]]),
             "line": p["line"],
             "line_name": p.get("line_name"),
             "year": int(p["year"]),
@@ -148,25 +154,43 @@ def route_units(pieces: list[dict[str, Any]]) -> list[dict[str, Any]]:
 def with_years(
     assignees: list[dict[str, Any]], units: list[dict[str, Any]]
 ) -> list[dict[str, Any]]:
-    """업체 줄에 연도 = 그 조각의 파일 연도(조각이 없으면 None) — 업체 줄 `year` 칸은 안 봄."""
-    years = {u["piece_id"]: u["year"] for u in units}
-    return [{**a, "year": years.get(a.get("piece_id"))} for a in assignees]
+    """업체 줄 열쇠 = 그 세트 열쇠(옛 조각 id 풀어 봄) · 연도 = 그 세트의 파일 연도(없으면 None) —
+    업체 줄 `year` 칸은 안 봄."""
+    out = []
+    for a in assignees:
+        unit = unit_of(a, units)
+        key = unit["piece_id"] if unit else a.get("piece_id")
+        out.append({**a, "piece_id": key, "year": unit and unit["year"]})
+    return out
 
 
 def unit_key(unit: dict[str, Any]) -> str | int:
-    """내보내기 묶음 열쇠 — 조각 id."""
+    """내보내기 묶음 열쇠 — 세트 열쇠."""
     return unit["piece_id"]
 
 
 def unit_label(unit: dict[str, Any]) -> str:
-    """zip 폴더 · 첨부 이름 — 「2026_갈래1_p3」."""
+    """zip 폴더 · 첨부 이름 — 「2027-1_갈래1」."""
     line = (unit["line_name"] or unit["line"]).replace(" ", "")
-    return f"{unit['year']}_{line}_{unit['piece_id']}"
+    return f"{unit['piece_id']}_{line}"
 
 
 def unit_of(assignee: dict[str, Any], units: list[dict[str, Any]]) -> dict[str, Any] | None:
-    """업체 행의 단위 = 그 조각."""
-    return next((u for u in units if u["piece_id"] == assignee.get("piece_id")), None)
+    """업체 행의 단위 = 그 세트(옛 조각 id = 그 조각이 든 세트)."""
+    return store.find_unit(units, assignee.get("piece_id"), "piece_id")
+
+
+def broken_sets(units: list[dict[str, Any]], years: list[int]) -> list[str]:
+    """확정 막을 세트 까닭 — 「2027-1 2 덩이」 · 「2027-2 갈래」."""
+    out = []
+    for unit in units:
+        if unit["year"] not in years:
+            continue
+        if unit["parts"] > 1:
+            out.append(f"{unit['piece_id']} {unit['parts']} 덩이")
+        if unit["branch"]:
+            out.append(f"{unit['piece_id']} 갈래")
+    return out
 
 
 def _saved(unit: dict[str, Any]) -> bool:
@@ -220,8 +244,10 @@ async def put_plan_assignees(
     _, units = await _units(project_id)
     piece_years = {u["piece_id"]: u["year"] for u in units}
     for item in request.assignees:
+        unit = store.find_unit(units, item.piece_id, "piece_id")  # 옛 조각 id = 든 세트
+        item.piece_id = unit["piece_id"] if unit else item.piece_id
         if item.piece_id not in piece_years:
-            raise HTTPException(status_code=400, detail=f"조각이 없습니다: {item.piece_id}")
+            raise HTTPException(status_code=400, detail=f"세트가 없습니다: {item.piece_id}")
     years = {piece_years[a.piece_id] for a in request.assignees} | set(request.years or [])
     unknown = sorted(years - set(piece_years.values()))
     if unknown:
@@ -267,12 +293,20 @@ def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]:
 async def finalize_plan(
     project_id: UUID, request: FinalizeRequest, session: dict = Depends(verify_session)
 ) -> dict[str, Any]:
-    """연도 노선 확정 → 지정 업체마다 그 조각 zip 을 첨부해 메일 → 보낸 곳에 발송 시각."""
+    """연도 노선 확정 → 지정 업체마다 그 세트 zip 을 첨부해 메일 → 보낸 곳에 발송 시각."""
     _require_forest(session)
     project = await _plan_project(project_id)
     rows, units = await _units(project_id)
     routes = unit_routes(units)
     saved = {r.year for r in routes}
+    asked = set(request.years) if request.years else {u["year"] for u in units}
+    broken = broken_sets(units, sorted(asked))
+    if broken:
+        raise HTTPException(
+            status_code=409,
+            detail=f"끊기거나 갈래가 있는 세트입니다: {', '.join(broken)}"
+            " — 예정노선에서 한 줄로 이어 오세요.",
+        )
     years = sorted(set(request.years) if request.years else saved)
     missing = [y for y in years if y not in saved]
     if not years or missing:
@@ -350,8 +384,19 @@ def year_summary(row: dict[str, Any]) -> dict[str, Any]:
 
 
 def unit_summary(unit: dict[str, Any]) -> dict[str, Any]:
-    """표 한 줄(조각) — label = 「본선 0+0~12+10」."""
-    out = {key: unit[key] for key in ("piece_id", "line", "line_name", "year", "length_m")}
+    """표 한 줄(세트) — label = 「2027-1」."""
+    fields = (
+        "piece_id",
+        "set",
+        "parts",
+        "branch",
+        "pieces",
+        "line",
+        "line_name",
+        "year",
+        "length_m",
+    )
+    out = {key: unit[key] for key in fields}
     out.update(saved=_saved(unit), confirmed_at=None, label=None)
     if unit["confirmed_at"]:
         out["confirmed_at"] = utc_text(unit["confirmed_at"])
@@ -364,9 +409,7 @@ def unit_summary(unit: dict[str, Any]) -> dict[str, Any]:
             end=_end_point(unit["planned"][-1], epsg),
         )
     if unit["piece_id"]:
-        out["label"] = piece_label(
-            unit["line_name"] or unit["line"], unit["start_chainage_m"], unit["end_chainage_m"]
-        )
+        out["label"] = unit["piece_id"]
     return out
 
 
@@ -379,7 +422,7 @@ def piece_label(line_name: str, from_m: float, to_m: float) -> str:
 async def get_plan_summary(
     project_id: UUID, session: dict = Depends(verify_session)
 ) -> dict[str, Any]:
-    """`years` = 연도 줄(확정 고르기) · `pieces` = 표 한 줄 단위(조각)."""
+    """`years` = 연도 줄(확정 고르기) · `pieces` = 표 한 줄 단위(세트)."""
     project = await _plan_project(project_id)
     rows, units = await _units(project_id)
     units, assignees = visible_to(session, units, await _assignees(project_id, units))
@@ -395,7 +438,7 @@ async def get_plan_summary(
 
 
 def export_zip(bundle: Any, units: list[dict[str, Any]], with_all: bool) -> bytes:
-    """단위마다 `<연도>_<선>_<조각>/` · 전체 `전체/` — 좌표표 이름이 같아 폴더로."""
+    """단위마다 `<세트>_<선>/` · 전체 `전체/` — 좌표표 이름이 같아 폴더로."""
     buffer = io.BytesIO()
     with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
         for unit in units:
@@ -409,7 +452,7 @@ def export_zip(bundle: Any, units: list[dict[str, Any]], with_all: bool) -> byte
 
 @router.get("/{project_id}/plan/export", response_model=None)
 async def export_plan(project_id: UUID, session: dict = Depends(verify_session)) -> Response:
-    """산림청 = 저장된 전 조각 + 전체 · 지정 회사 = 확정된 자기 조각만(전체 없음)."""
+    """산림청 = 저장된 전 세트 + 전체 · 지정 회사 = 확정된 자기 세트만(전체 없음)."""
     project = await _plan_project(project_id)
     _, units = await _units(project_id)
     routes = unit_routes(units)
@@ -421,7 +464,7 @@ async def export_plan(project_id: UUID, session: dict = Depends(verify_session))
         own, _ = visible_to(session, picked, await _assignees(project_id, units))
         picked = [u for u in own if u["confirmed_at"]]
         if not picked:
-            raise HTTPException(status_code=403, detail="받을 수 있는 조각이 없습니다.")
+            raise HTTPException(status_code=403, detail="받을 수 있는 세트가 없습니다.")
     first = float(project.get("route_start_m") or 0.0)
     bundle = await asyncio.to_thread(
         build_export, project["name"], project["region"] or "", routes, first

resources/tester/test_530_2_plan_analyze.py

@@ -210,7 +210,7 @@ def test_put_writes_file_and_year_status_only(client) -> None:
     assert got.status_code == 200 and got.json()["derived"]["pieces"][1]["from_m"] == 75.0
     assert got.json()["year_start"] == 2026
     saved = json.loads(client.file.read_text("utf-8"))
-    assert saved["version"] == 1 and saved["epsg"] == 5186
+    assert saved["version"] == 2 and saved["epsg"] == 5186
     assert saved["points"] == BASE["points"]  # 몸에 지점 없음 = 그대로
     assert [p["id"] for p in saved["pieces"]] == ["p1", "p2", "p3"]
     assert "length_m" not in json.dumps(saved)  # 파생 저장 안 함
@@ -228,7 +228,11 @@ def test_put_writes_file_and_year_status_only(client) -> None:
 
 def test_put_reset_drops_year_rows(client) -> None:
     got = client.put(URL, json={"epsg": 5186, "lines": [], "marks": [], "pieces": []})
-    assert got.status_code == 200 and got.json()["derived"] == {"lines": [], "pieces": []}
+    assert got.status_code == 200 and got.json()["derived"] == {
+        "lines": [],
+        "pieces": [],
+        "sets": [],
+    }
     sqls = [s for s, _ in client.db.log]
     assert "DELETE FROM plan_route_years WHERE project_id = %s" in sqls
     assert not any("project_year_end" in s for s in sqls)

resources/tester/test_531_3_plan_link.py

@@ -64,6 +64,9 @@ def _assignee(**kw):
         "project_id": PLAN_ID,
         "year": 2026,
         "piece_id": "p26",
+        "set_key": "2026-1",  # 638-1 — 옛 조각 id p26 이 든 세트
+        "parts": 1,
+        "branch": False,
         "company_id": 5,
         "linked_project_id": None,
         "name": "시험 예정노선",
@@ -400,13 +403,13 @@ def _zip_names(response):
 def test_export_forest_gets_all(plan_repo):
     forest = {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1}
     names = _zip_names(asyncio.run(d04.export_plan(PLAN_ID, forest)))
-    assert {n.split("/")[0] for n in names} == {"2026_본선_p26", "2027_본선_p27", "전체"}
+    assert {n.split("/")[0] for n in names} == {"2026-1_본선", "2027-1_본선", "전체"}
     assert sum(n.endswith(".shp") for n in names) == 3
 
 
 def test_export_shared_gets_own_confirmed_year(plan_repo):
     names = _zip_names(asyncio.run(d04.export_plan(PLAN_ID, SESSION)))
-    assert {n.split("/")[0] for n in names} == {"2026_본선_p26"}  # 2027 = 확정 전 · 전체 없음
+    assert {n.split("/")[0] for n in names} == {"2026-1_본선"}  # 2027 = 확정 전 · 전체 없음
     with pytest.raises(HTTPException) as err:
         asyncio.run(d04.export_plan(PLAN_ID, {**SESSION, "company_id": 9}))
     assert err.value.status_code == 403

resources/tester/test_531_d04_export_finalize.py

@@ -349,10 +349,10 @@ def test_finalize_sends_zip_per_assignee(fake_db):
     assert to == "ga@x.kr" and "2026" in subject
     assert set(fake_db["confirmed"]) == {2026, 2027}  # 연도 → 파일 지문
     name, blob, mime = attachments[0]
-    assert name == "시험_예정노선_2026_본선_p26.zip" and mime == "application/zip"
+    assert name == "시험_예정노선_2026-1_본선.zip" and mime == "application/zip"  # 638-1 세트
     assert len(blob) > 1000
     with zipfile.ZipFile(io.BytesIO(blob)) as archive:
-        assert any(n.endswith("_2026_본선_p26.shp") for n in archive.namelist())
+        assert any(n.endswith("_2026_본선_2026-1.shp") for n in archive.namelist())
         assert not any("_2027" in n for n in archive.namelist())  # 다른 연도는 안 샘
 
     # 다시 보내기

resources/tester/test_563_2_plan_pieces.py

@@ -124,17 +124,18 @@ def _names(response) -> list[str]:
 def test_summary_piece_rows_and_firm_view(plan):
     out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))
     rows = out["pieces"]
-    assert [(r["piece_id"], r["year"], r["label"]) for r in rows] == [
-        ("p1", 2026, "본선 0+0~12+10"),
-        ("p3", 2026, "갈래 1 0+0~6+10"),
-        ("p2", 2027, "본선 12+10~25+19.3"),
+    # 638-1 — 단위 = 세트(옛 조각 하나 = 세트 하나 · 그 해 선 차례 · 거리 차례 번호)
+    assert [(r["piece_id"], r["year"], r["label"], r["pieces"]) for r in rows] == [
+        ("2026-1", 2026, "2026-1", ["p1"]),
+        ("2026-2", 2026, "2026-2", ["p3"]),
+        ("2027-1", 2027, "2027-1", ["p2"]),
     ]
     assert rows[1]["length_m"] == 130.0 and rows[1]["start"]["x"] == P3[0][0]
     assert rows[0]["confirmed_at"] == "x" and rows[2]["confirmed_at"] is None
     assert [y["year"] for y in out["years"]] == [2026, 2027]
     assert [(a["id"], a["year"]) for a in out["assignees"]] == [(7, 2026), (8, 2026)]
     # 같은 연도 다른 업체 — 자기 조각 · 자기 행만
-    for session, piece, aid in ((FIRM5, "p1", 7), (FIRM6, "p3", 8)):
+    for session, piece, aid in ((FIRM5, "2026-1", 7), (FIRM6, "2026-2", 8)):
         mine = asyncio.run(d04.get_plan_summary(PLAN_ID, session))
         assert [r["piece_id"] for r in mine["pieces"]] == [piece]
         assert [a["id"] for a in mine["assignees"]] == [aid]
@@ -145,9 +146,9 @@ def test_summary_piece_rows_and_firm_view(plan):
 def test_export_piece_folders_and_firm_only_own(plan):
     names = _names(asyncio.run(d04.export_plan(PLAN_ID, FOREST)))
     assert {n.split("/")[0] for n in names} == {
-        "2026_본선_p1",
-        "2026_갈래1_p3",
-        "2027_본선_p2",
+        "2026-1_본선",
+        "2026-2_갈래1",
+        "2027-1_본선",
         "전체",
     }
     assert sum(n.endswith(".shp") for n in names) == 4
@@ -164,7 +165,7 @@ def test_export_piece_folders_and_firm_only_own(plan):
         round(250.0 + L2, 1),
     ]
     firm = _names(asyncio.run(d04.export_plan(PLAN_ID, FIRM6)))
-    assert {n.split("/")[0] for n in firm} == {"2026_갈래1_p3"}
+    assert {n.split("/")[0] for n in firm} == {"2026-2_갈래1"}
     plan["file"]["confirmed"].clear()  # 확정 전 = 받을 것 없음
     with pytest.raises(HTTPException) as err:
         asyncio.run(d04.export_plan(PLAN_ID, FIRM6))
@@ -174,14 +175,15 @@ def test_export_piece_folders_and_firm_only_own(plan):
 def test_finalize_mails_own_piece_zip(plan):
     out = asyncio.run(d04.finalize_plan(PLAN_ID, d04.FinalizeRequest(years=[2026]), FOREST))
     assert [(n["piece_id"], n["state"]) for n in out["notifications"]] == [
-        ("p1", "sent"),
-        ("p3", "sent"),
+        ("2026-1", "sent"),
+        ("2026-2", "sent"),
     ]
-    for (to, attachments), piece in zip(plan["mails"], ("본선_p1", "갈래1_p3"), strict=True):
+    pairs = (("2026-1", "본선"), ("2026-2", "갈래1"))
+    for (to, attachments), (key, line) in zip(plan["mails"], pairs, strict=True):
         name, blob, _ = attachments[0]
-        assert name == f"시험_예정노선_2026_{piece}.zip"
+        assert name == f"시험_예정노선_{key}_{line}.zip"
         shp = [n for n in zipfile.ZipFile(io.BytesIO(blob)).namelist() if n.endswith(".shp")]
-        assert len(shp) == 1 and shp[0].endswith(f"_2026_{piece}.shp")
+        assert len(shp) == 1 and shp[0].endswith(f"_2026_{line}_{key}.shp")
 
 
 def test_put_assignees_by_piece(plan):
@@ -202,7 +204,9 @@ def test_put_assignees_by_piece(plan):
     )
     deleted, updates, inserts = plan["applied"]
     assert deleted == [8] and [u["id"] for u in updates] == [7]
-    assert [(i["piece_id"], i["year"], i["company_id"]) for i in inserts] == [("p3", 2026, 5)]
+    # 옛 조각 id 로 보내도 세트 열쇠로 적음(638-1)
+    assert [(i["piece_id"], i["year"], i["company_id"]) for i in inserts] == [("2026-2", 2026, 5)]
+    assert [u["piece_id"] for u in updates] == ["2026-1"]
 
 
 def test_confirm_drops_when_file_changes(plan, monkeypatch):
@@ -280,7 +284,7 @@ def test_assigned_list_piece_label_length(plan, monkeypatch):
     monkeypatch.setattr(d01, "get_db_pool", lambda: _Conn(rows))
     out = asyncio.run(d01.assigned_plan_routes(FIRM6))["routes"]
     assert [(r["piece_id"], r["year"], r["piece_label"], r["length_m"]) for r in out] == [
-        ("p3", 2026, "갈래 1 0+0~6+10", 130.0),
+        ("2026-2", 2026, "2026-2", 130.0),
     ]
 
 
@@ -289,17 +293,17 @@ def test_line_name_flows_to_labels_zip_xlsx(plan):
     named = [{**LINES[0], "name": "용화 본선"}, LINES[1]]
     plan["file"]["plan"] = plan_of(named, MARKS, PIECES)
     rows = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))["pieces"]
-    assert [r["label"] for r in rows] == [
-        "용화 본선 0+0~12+10",
-        "갈래 1 0+0~6+10",
-        "용화 본선 12+10~25+19.3",
+    assert [(r["label"], r["line_name"]) for r in rows] == [
+        ("2026-1", "용화 본선"),
+        ("2026-2", "갈래 1"),
+        ("2027-1", "용화 본선"),
     ]
     body = asyncio.run(d04.export_plan(PLAN_ID, FOREST)).body
     with zipfile.ZipFile(io.BytesIO(body)) as z:
         assert {n.split("/")[0] for n in z.namelist()} == {
-            "2026_용화본선_p1",
-            "2026_갈래1_p3",
-            "2027_용화본선_p2",
+            "2026-1_용화본선",
+            "2026-2_갈래1",
+            "2027-1_용화본선",
             "전체",
         }
         book = next(n for n in z.namelist() if n.startswith("전체/") and n.endswith(".xlsx"))

resources/tester/test_590_1_plan_file.py

@@ -105,7 +105,7 @@ def test_atomic_write_keeps_old_file_on_failure(tmp_path, monkeypatch):
     path = store.plan_path(tmp_path)
     assert (
         path.parent.name == "D03_PlanAnalysis"
-        and json.loads(path.read_text("utf-8"))["version"] == 1
+        and json.loads(path.read_text("utf-8"))["version"] == 2  # 638-1 판 2(조각 set)
     )
     assert store.read_plan(tmp_path) == plan
 
@@ -124,7 +124,7 @@ def test_atomic_write_keeps_old_file_on_failure(tmp_path, monkeypatch):
 def test_newer_version_refused(tmp_path):
     path = store.plan_path(tmp_path)
     path.parent.mkdir(parents=True)
-    path.write_text(json.dumps({**store.empty_plan(), "version": 2}), "utf-8")
+    path.write_text(json.dumps({**store.empty_plan(), "version": 3}), "utf-8")
     with pytest.raises(HTTPException) as err:
         store.read_plan(tmp_path)
     assert err.value.status_code == 409
@@ -170,10 +170,10 @@ def test_two_saves_do_not_interleave(monkeypatch, tmp_path):
     async def project_info(_pid):
         return {"year_start": 2026, "site": None}
 
-    async def slow_status(connection, project_id, pieces, years):
-        order.append(f"start {len(pieces)}")
+    async def slow_status(connection, project_id, keys, years, before=None):
+        order.append(f"start {len(keys)}")
         await asyncio.sleep(0.05)
-        order.append(f"end {len(pieces)}")
+        order.append(f"end {len(keys)}")
 
     class Conn:
         async def __aenter__(self):

resources/tester/test_638_sets.py

@@ -0,0 +1,263 @@
+# -*- coding: utf-8 -*-
+"""638-1 예정노선 노선 세트(연도-번호 · 요구 547 · 548) — 서버.
+
+- 조각 `set` · 옛 조각(set 없음) = 그 해 선 차례 · 거리 차례로 최댓값 + 1
+- 같은 선 이웃 같은 세트 = 하나(앞 조각 id · 사이 마크 걷음)
+- 세트 셈 `derived.sets[]` — 덩이 수 · 갈래 · 길이 · 조각
+- 업체 줄 열쇠 = 세트 · 옛 조각 id = 든 세트 · 사라진 세트 409 · 옛 열쇠 → 세트 열쇠
+- D04 확정 막음(덩이 2 이상 · 갈래) · 연결 = 세트 한 줄(아니면 거절) · 파일 판 2
+- 거울 사례(`fixtures/d03_638_set_cases.json`) = TS `logic/D03_PlanAnalysis_Group.ts` 와 같은 답
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from helper_plan_file import plan_of, use_plan  # noqa: E402
+
+from D03_PlanAnalysis import D03_PlanAnalysis_Network as net  # noqa: E402
+from D03_PlanAnalysis import D03_PlanAnalysis_Store as store  # noqa: E402
+from D04_PlanConfirm import D04_PlanConfirm_Link as link  # noqa: E402
+from D04_PlanConfirm import D04_PlanConfirm_Repository as repo  # noqa: E402
+from D04_PlanConfirm import D04_PlanConfirm_Router as d04  # noqa: E402
+
+PLAN_ID = str(uuid4())
+FOREST = {"role": "SYSTEM_ADMIN", "email": "root@x.kr", "company_id": 1, "user_id": 1}
+MAIN = {"id": "main", "vertices": [[0.0, 0.0], [100.0, 0.0], [200.0, 0.0], [300.0, 0.0]]}
+
+
+def _piece(pid, a, b, year=2027, no=None, line="main"):
+    out = {"id": pid, "line": line, "from_m": a, "to_m": b, "year": year}
+    return {**out, "set": no} if no else out
+
+
+def _sets(plan):
+    return [
+        (s["key"], s["parts"], s["branch"], round(s["length_m"], 3), s["pieces"])
+        for s in net.derive(plan)["sets"]
+    ]
+
+
+def test_old_pieces_numbered_by_line_then_distance():
+    plan = plan_of(
+        [MAIN],
+        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
+        [_piece("p3", 200.0, 300.0), _piece("p1", 0.0, 100.0, 2028), _piece("p2", 100.0, 200.0)],
+    )
+    assert _sets(plan) == [
+        ("2027-1", 1, False, 100.0, ["p2"]),
+        ("2027-2", 1, False, 100.0, ["p3"]),
+        ("2028-1", 1, False, 100.0, ["p1"]),
+    ]
+    # 있는 번호 뒤로(새 조각 = 그 해 최댓값 + 1)
+    plan["pieces"][1]["year"] = 2027
+    plan["pieces"][2]["set"] = 5
+    keys = {p["id"]: p["set_key"] for p in net.derive(plan)["pieces"]}
+    assert keys == {"p1": "2027-6", "p2": "2027-5", "p3": "2027-7"}
+
+
+def test_same_set_neighbours_merge_and_drop_mark():
+    plan = plan_of(
+        [MAIN],
+        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
+        [
+            _piece("a", 0.0, 100.0, no=1),
+            _piece("b", 100.0, 200.0, no=1),
+            _piece("c", 200, 300, no=2),
+        ],
+    )
+    flat = net.normalize(plan)
+    assert flat["marks"] == [{"line": "main", "at_m": 200.0}]
+    assert [(p["id"], p["from_m"], p["to_m"], p["set"]) for p in flat["pieces"]] == [
+        ("a", 0.0, 200.0, 1),
+        ("c", 200, 300, 2),
+    ]
+    assert _sets(plan) == [("2027-1", 1, False, 200.0, ["a"]), ("2027-2", 1, False, 100.0, ["c"])]
+    assert net.normalize(flat) == flat  # 두 번 해도 같음
+    # 다른 연도 같은 번호 = 안 합침
+    plan["pieces"][1]["year"] = 2028
+    assert len(net.normalize(plan)["pieces"]) == 3
+
+
+def test_parts_branch_and_join():
+    # 같은 세트가 다른 세트를 사이에 두고 끊김 → 2 덩이
+    gap = plan_of(
+        [MAIN],
+        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
+        [_piece("a", 0, 100, no=1), _piece("b", 100, 200, no=2), _piece("c", 200, 300, no=1)],
+    )
+    assert _sets(gap)[0] == ("2027-1", 2, False, 200.0, ["a", "c"])
+    # 본선 가운데 꼭짓점에서 갈래 · 같은 세트 → 갈래
+    fork = plan_of(
+        [
+            MAIN,
+            {
+                "id": "b",
+                "vertices": [[100.0, 0.0], [100.0, 50.0]],
+                "branch": {"line": "main", "vertex": 1},
+            },
+        ],
+        [],
+        [_piece("m", 0, 300, no=1), _piece("g", 0, 50, no=1, line="b")],
+    )
+    assert _sets(fork) == [("2027-1", 1, True, 350.0, ["m", "g"])]
+    # 끝끼리 닿은 다른 선(뒤집힌 방향) = 한 덩이 · 이은 한 줄
+    tail = {"id": "t", "vertices": [[400.0, 0.0], [300.0, 0.0]]}
+    chain = plan_of(
+        [MAIN, tail], [], [_piece("m", 0, 300, no=1), _piece("t", 0, 100, no=1, line="t")]
+    )
+    assert _sets(chain) == [("2027-1", 1, False, 400.0, ["m", "t"])]
+    _, units = store.plan_parts(chain)
+    assert units[0]["planned"] == [[0, 0], [100, 0], [200, 0], [300, 0], [400, 0]]
+    assert (units[0]["from_m"], units[0]["to_m"]) == (0.0, 400.0)
+    # 끊긴 세트 · 갈래 세트 = 이은 줄 없음
+    assert store.plan_parts(gap)[1][0]["planned"] == []
+    assert store.plan_parts(fork)[1][0]["planned"] == []
+
+
+def test_find_unit_old_piece_key():
+    plan = plan_of(
+        [MAIN], [{"line": "main", "at_m": 100.0}], [_piece("p1", 0, 100), _piece("p2", 100, 300)]
+    )
+    _, units = store.plan_parts(plan)
+    assert [u["id"] for u in units] == ["2027-1", "2027-2"]
+    assert store.find_unit(units, "2027-2")["pieces"] == ["p2"]
+    assert store.find_unit(units, "p2")["id"] == "2027-2"  # 옛 업체 열쇠
+    assert store.find_unit(units, "없음") is None
+    assert net.old_keys(plan) == {"p1": "2027-1", "p2": "2027-2"}
+
+
+def test_file_version_two(tmp_path):
+    assert store.VERSION == 2
+    plan = plan_of([MAIN], [], [_piece("p1", 0, 300, no=1)])
+    store.save_plan(tmp_path, plan)
+    saved = json.loads(store.plan_path(tmp_path).read_text("utf-8"))
+    assert saved["version"] == 2 and saved["pieces"][0]["set"] == 1
+    # 판 1(set 없음) = 읽힘 · 보기에 번호
+    store.plan_path(tmp_path).write_text(
+        json.dumps({**saved, "version": 1, "pieces": [_piece("p1", 0, 300)]}), "utf-8"
+    )
+    old = store.read_plan(tmp_path)
+    view = net._view(old, {"site": None, "year_start": 2027})
+    assert (
+        view["network"]["pieces"][0]["set"] == 1 and view["derived"]["sets"][0]["key"] == "2027-1"
+    )
+    assert net.Piece(**_piece("p1", 0, 1)).set is None  # 옛 화면 몸 = 받음
+    with pytest.raises(ValueError):
+        net.Piece(**_piece("p1", 0, 1), set=0)
+
+
+class _Cursor:
+    def __init__(self, held):
+        self.held, self.log, self.rows = held, [], []
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+    async def execute(self, sql, args=()):
+        self.log.append((sql, args))
+        self.rows = [(k,) for k in self.held] if sql.startswith("SELECT DISTINCT") else []
+
+    async def fetchall(self):
+        return self.rows
+
+
+class _Conn:
+    def __init__(self, held):
+        self.c = _Cursor(held)
+
+    def cursor(self):
+        return self.c
+
+
+def test_store_status_lost_set_and_old_key_rewrite():
+    conn = _Conn({"p1", "2027-2"})
+    years = {2027: {"fingerprint": "f"}}
+    asyncio.run(net._store_status(conn, PLAN_ID, {"2027-1", "2027-2"}, years, {"p1": "2027-1"}))
+    moved = [a for s, a in conn.c.log if s.startswith("UPDATE plan_route_assignees")]
+    assert moved == [("2027-1", PLAN_ID, "p1")]
+    conn = _Conn({"p1", "2027-9"})
+    with pytest.raises(HTTPException) as err:
+        asyncio.run(net._store_status(conn, PLAN_ID, {"2027-1"}, years, {"p1": "2027-1"}))
+    assert err.value.status_code == 409 and "2027-9" in err.value.detail
+
+
+@pytest.fixture
+def plan(monkeypatch):
+    gap = plan_of(
+        [MAIN],
+        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
+        [_piece("a", 0, 100, no=1), _piece("b", 100, 200, no=2), _piece("c", 200, 300, no=1)],
+    )
+    state = use_plan(monkeypatch, gap)
+
+    async def project(_pid):
+        return {"id": PLAN_ID, "name": "시험", "region": "용화", "project_kind": "PLAN"}
+
+    async def none(*a):
+        return []
+
+    monkeypatch.setattr(repo, "get_plan_project", project)
+    monkeypatch.setattr(repo, "list_assignees", none)
+    monkeypatch.setattr(repo, "mark_confirmed", none)
+    return state
+
+
+def test_d04_units_and_finalize_blocked(plan):
+    out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))
+    assert [(r["piece_id"], r["parts"], r["saved"]) for r in out["pieces"]] == [
+        ("2027-1", 2, False),
+        ("2027-2", 1, True),
+    ]
+    with pytest.raises(HTTPException) as err:
+        asyncio.run(d04.finalize_plan(PLAN_ID, d04.FinalizeRequest(years=[2027]), FOREST))
+    assert err.value.status_code == 409 and "2027-1 2 덩이" in err.value.detail
+    assert "예정노선에서" in err.value.detail
+
+
+def test_link_needs_one_line(plan, monkeypatch):
+    route = asyncio.run(link.piece_route(PLAN_ID, "a"))  # 옛 조각 id → 세트
+    assert (route["set_key"], route["parts"], route["route"]["planned"]) == ("2027-1", 2, [])
+
+    async def load(_id):
+        return {"id": 7, "company_id": 5, "project_id": PLAN_ID, "piece_id": "a", **route}
+
+    monkeypatch.setattr(link, "load_assignee", load)
+    payload = link.LinkRequest(mode="new")
+    with pytest.raises(HTTPException) as err:
+        asyncio.run(link.link_plan_route(7, payload, {"company_id": 5, "user_id": 3}))
+    assert (
+        err.value.status_code == 409
+        and "2027-1" in err.value.detail
+        and "2 덩이" in err.value.detail
+    )
+    one = asyncio.run(link.piece_route(PLAN_ID, "2027-2"))
+    assert one["route"]["planned"] == [[100.0, 0.0], [200.0, 0.0]] and one["parts"] == 1
+
+
+CASES = ROOT / "resources" / "tester" / "fixtures" / "d03_638_set_cases.json"
+
+
+@pytest.mark.skipif(not CASES.is_file(), reason="거울 사례 파일 = sub6 638-2 가 만듦")
+def test_mirror_cases_match_ts():
+    cases = json.loads(CASES.read_text("utf-8"))
+    assert cases["cases"]
+    for case in cases["cases"]:
+        net.derive(case["network"], cases["year_start"])  # 틀린 사례 없음
+        got = net.normalize(case["network"])["pieces"]
+        keys = ("id", "line", "from_m", "to_m", "year", "set")
+        assert [{k: p[k] for k in keys} for p in got] == case["expect"], case["name"]