feat(563-2): D04 확정 · 업체 지정 · 내보내기 · 연결 = 조각 단위 — 옛 연도 행 프로젝트 그대로

- 단위 = plan_route_pieces 한 줄 · 조각 없는 연도 행 = 연도 하나 · 확정은 연도 단위 그대로
- 요약 pieces(이름 「본선 0+0~12+10」) · 업체 행 piece_id · 지정 회사 = 자기 조각만
- 내보내기 폴더 = 연도_선_조각 · 전체 = 선마다 레코드 · 좌표표 선 · 조각 칸
- 확정 메일 = 업체마다 자기 조각 zip · 연결 · 가져오기 = 조각 꺾은선
- 받은 예정노선 목록 = 조각 이름 · 길이
- 시험 531 둘 = 조각 없음 가짜 더함 · 새 563_2 지킴 시험

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

D01_PlanRegister/D01_PlanRegister_Router.py

@@ -4,6 +4,7 @@
 = 설계속도 작업임도 20 · 측점 없음(DB 칸 그대로).
 """
 
+import json
 from typing import Any
 
 import aiomysql
@@ -72,8 +73,11 @@ async def post_plan_project(
 async def assigned_plan_routes(
     session: dict[str, Any] = Depends(require_company),
 ) -> dict[str, Any]:
-    """우리 회사에 지정된 예정노선 연도 — 확정된 것(또는 이미 연결한 것) · 연결 상태 · 업데이트."""
+    """우리 회사에 지정된 예정노선 조각(옛 = 연도) — 확정된 것(또는 이미 연결한 것) · 연결 상태 ·
+    업데이트 · 조각 이름 · 길이(조각 행 · 옛 = 연도 행)."""
+    from D04_PlanConfirm import D04_PlanConfirm_Repository as plan_repo
     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,
@@ -81,23 +85,43 @@ async def assigned_plan_routes(
     ):
         await cursor.execute(
             """SELECT a.id AS assignee_id, a.project_id, p.name AS project_name, a.year,
+                      a.piece_id, c.length_m AS piece_length_m,
                       l.id AS linked_project_id, l.name AS linked_project_name,
-                      l.source_fingerprint, y.route_json, y.confirmed_at
+                      l.source_fingerprint, COALESCE(c.route_json, y.route_json) AS route_json,
+                      y.confirmed_at
                FROM plan_route_assignees a
                JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL
                LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year
+               LEFT JOIN plan_route_pieces c
+                 ON c.project_id = a.project_id AND c.piece_id = a.piece_id
                LEFT JOIN projects l ON l.id = a.linked_project_id AND l.deleted_at IS NULL
                WHERE a.company_id = %s
                  AND (y.confirmed_at IS NOT NULL OR l.id IS NOT NULL)
-               ORDER BY p.name, a.year""",
+               ORDER BY p.name, a.year, a.piece_id""",
             (session.get("company_id"),),
         )
         rows = await cursor.fetchall()
+    pieces: dict[tuple[str, str], dict[str, Any]] = {}
+    for project_id in {r["project_id"] for r in rows if r.get("piece_id")}:
+        for piece in await plan_repo.list_pieces(str(project_id)):
+            pieces[(str(project_id), piece["piece_id"])] = piece
     routes = []
     for row in rows:
         route_json, fingerprint = row.pop("route_json"), row.pop("source_fingerprint")
+        piece_length = row.pop("piece_length_m", None)
         confirmed = row["confirmed_at"] is not None
+        body = json.loads(route_json) if isinstance(route_json, str | bytes) else route_json
+        piece = pieces.get((str(row["project_id"]), row.get("piece_id")))
         routes.append(
-            {**row, "update_available": update_available(route_json, fingerprint, confirmed)}
+            {
+                **row,
+                "piece_id": row.get("piece_id"),
+                "piece_label": piece
+                and piece_label(piece["line_name"], piece["from_m"], piece["to_m"]),
+                "length_m": piece_length
+                if piece_length is not None
+                else (body or {}).get("length_m"),
+                "update_available": update_available(route_json, fingerprint, confirmed),
+            }
         )
     return {"status": "success", "routes": routes}

D04_PlanConfirm/D04_PlanConfirm_Export.py

@@ -1,4 +1,5 @@
-"""D04 예정노선 내보내기 — 연도별 노선(꺾은선)을 용화 예정노선 양식 shapefile 한 벌로 만든다.
+"""D04 예정노선 내보내기 — 조각(구역 × 갈래)마다 · 옛 프로젝트는 연도마다 노선(꺾은선)을
+용화 예정노선 양식 shapefile 한 벌로 만든다.
 
 양식 = `resources/knowledge/original/실무문서/2026년 임도 용화/예정노선/대상지_일월.용화.2.2.*`
 (분석 345-C): 2D 폴리라인 1레코드 · UTM-K(EPSG 5179) · cp949 · 칸 `대상지` C(20) ·
@@ -70,9 +71,10 @@ def station_label(chainage_m: float, interval_m: float = STATION_INTERVAL_M) ->
 
 @dataclass
 class YearRoute:
-    """한 연도의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점).
+    """한 조각(또는 옛 연도)의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점).
 
-    누가거리 두 값이 있으면 그대로 칸에 쓰고(D03 이은 값), 없으면 연도 순서로 이어 셈.
+    누가거리 두 값이 있으면 그대로 칸에 쓰고(조각 = 선마다 No.0), 없으면 연도 순서로 이어 셈.
+    `piece` = 조각 id(옛 연도 노선은 None) · `line` = 선 이름(파일 이름 · 전체 묶음 열쇠).
     """
 
     year: int
@@ -80,11 +82,18 @@ class YearRoute:
     crs_epsg: int
     start_station_m: float | None = None
     end_station_m: float | None = None
+    piece: str | None = None
+    line: str | None = None
+
+    @property
+    def key(self) -> str | int:
+        return self.piece or self.year
 
 
 @dataclass
 class ExportBundle:
-    """`by_year[연도]` = 그 연도 shp 한 벌 + 좌표표 · `all` = 전 연도 이은 한 벌 + 좌표표."""
+    """`by_year[열쇠]` = 그 조각(열쇠 = 조각 id · 옛은 연도) shp 한 벌 + 좌표표 ·
+    `all` = 선마다 이은 레코드 한 벌 + 좌표표."""
 
     by_year: dict[int, dict[str, bytes]] = field(default_factory=dict)
     all: dict[str, bytes] = field(default_factory=dict)
@@ -120,20 +129,22 @@ def _target_text(region: str, length_m: float) -> str:
     return f"{region}.{length_m / 1000:.1f}"
 
 
-def _write_shapefile(
-    points: list[tuple[float, float]], year: int | None, stations: tuple[float, float], region: str
-) -> dict[str, bytes]:
-    """5179 꺾은선 → `.shp/.shx/.dbf/.prj/.cpg` 바이트 (파일명은 호출부가 붙임: 키 = 확장자)."""
-    length_m = polyline_length_m(points)
+Record = tuple[list[tuple[float, float]], int | None, tuple[float, float]]
+
+
+def _write_shapefile(records: list[Record], region: str) -> dict[str, bytes]:
+    """5179 꺾은선 레코드(점 · 연도 · 누가 두 값) → `.shp/.shx/.dbf/.prj/.cpg` 바이트
+    (파일명은 호출부가 붙임: 키 = 확장자)."""
+    lengths = [polyline_length_m(points) for points, _, _ in records]
     frame = gpd.GeoDataFrame(
         {
-            "대상지": [_clip_bytes(_target_text(region, length_m))],
-            "길이": [round(length_m / 1000, 3)],
-            "연도": [year],
-            "시점누가": [round(stations[0], 2)],
-            "종점누가": [round(stations[1], 2)],
+            "대상지": [_clip_bytes(_target_text(region, m)) for m in lengths],
+            "길이": [round(m / 1000, 3) for m in lengths],
+            "연도": [year for _, year, _ in records],
+            "시점누가": [round(stations[0], 2) for _, _, stations in records],
+            "종점누가": [round(stations[1], 2) for _, _, stations in records],
         },
-        geometry=[LineString(points)],
+        geometry=[LineString(points) for points, _, _ in records],
         crs=EXPORT_EPSG,
     )
     with tempfile.TemporaryDirectory() as tmp:
@@ -169,6 +180,8 @@ def _coordinate_xlsx(project_name: str, rows: list[dict]) -> bytes:
             "시점누가(m)",
             "종점누가(m)",
             "연장(m)",
+            "선",
+            "조각",
         ]
     )
     for row in rows:
@@ -190,6 +203,8 @@ def _coordinate_xlsx(project_name: str, rows: list[dict]) -> bytes:
                 round(row["start_station"], 2),
                 round(row["end_station"], 2),
                 round(row["length_m"], 2),
+                row["line"],
+                row["piece"],
             ]
         )
     buffer = io.BytesIO()
@@ -208,10 +223,10 @@ def build_export(
     first_station_m: float = 0.0,
     with_xlsx: bool = True,
 ) -> ExportBundle:
-    """연도 노선들 → 연도별 한 벌 + 전체 한 벌(+ 좌표표).
+    """조각(옛 = 연도) 노선들 → 조각마다 한 벌 + 전체 한 벌(+ 좌표표).
 
-    누가거리 = `first_station_m` 에서 시작해 연도 순서로 이어 셈(직전 연도 종점 = 다음 연도 시점).
-    shp 안 측점은 파일마다 0 부터가 아니라 이 누가값을 칸으로만 보관(B03 은 칸을 무시).
+    누가거리 = 값이 없으면 `first_station_m` 에서 시작해 연도 순서로 이어 셈(직전 연도 종점 =
+    다음 연도 시점). shp 안 측점은 칸으로만 보관(B03 은 칸을 무시).
     """
     ordered = sorted(years, key=lambda y: y.year)
     rows: list[dict] = []
@@ -232,6 +247,9 @@ def build_export(
                 "length_m": length_m,
                 "start_station": start,
                 "end_station": end,
+                "key": item.key,
+                "piece": item.piece,
+                "line": item.line,
             }
         )
         station = end
@@ -241,39 +259,40 @@ def build_export(
     xlsx_name = f"좌표표_{safe_name(project_name)}.xlsx"
     for row in rows:
         base = f"대상지_{region_name}.{row['length_m'] / 1000:.1f}_{row['year']}"
-        files = _named(
-            base,
-            _write_shapefile(
-                row["points"], row["year"], (row["start_station"], row["end_station"]), region
-            ),
-        )
+        if row["piece"]:
+            base += f"_{safe_name(row['line'] or '')}_{safe_name(row['piece'])}"
+        stations = (row["start_station"], row["end_station"])
+        files = _named(base, _write_shapefile([(row["points"], row["year"], stations)], region))
         if with_xlsx:
             files[xlsx_name] = _coordinate_xlsx(project_name, [row])
-        bundle.by_year[row["year"]] = files
-        bundle.lengths_m[row["year"]] = row["length_m"]
+        bundle.by_year[row["key"]] = files
+        bundle.lengths_m[row["key"]] = row["length_m"]
 
-    # 전체: 앞 연도 종점 = 다음 연도 시점이라 겹치는 점 하나는 뺌.
-    merged: list[tuple[float, float]] = []
+    # 전체: 선마다 한 레코드 · 앞 조각 종점 = 다음 조각 시점이라 겹치는 점 하나는 뺌.
+    by_line: dict[str | None, list[dict]] = {}
     for row in rows:
-        merged.extend(
-            row["points"][1:] if merged and merged[-1] == row["points"][0] else row["points"]
-        )
-    total_m = polyline_length_m(merged)
+        by_line.setdefault(row["line"], []).append(row)
+    records: list[Record] = []
+    for parts in by_line.values():
+        parts.sort(key=lambda r: r["start_station"])
+        merged: list[tuple[float, float]] = []
+        for row in parts:
+            same = merged and math.dist(merged[-1], row["points"][0]) < 1e-6
+            merged.extend(row["points"][1:] if same else row["points"])
+        records.append((merged, None, (parts[0]["start_station"], parts[-1]["end_station"])))
+    total_m = sum(polyline_length_m(points) for points, _, _ in records)
     bundle.lengths_m[0] = total_m
     base = f"대상지_{region_name}.{total_m / 1000:.1f}_전체"
-    bundle.all = _named(
-        base,
-        _write_shapefile(merged, None, (rows[0]["start_station"], rows[-1]["end_station"]), region),
-    )
+    bundle.all = _named(base, _write_shapefile(records, region))
     if with_xlsx:
         bundle.all[xlsx_name] = _coordinate_xlsx(project_name, rows)
     return bundle
 
 
-def company_zip(project_name: str, year: int, files: dict[str, bytes]) -> tuple[str, bytes]:
-    """업체에 보낼 `<프로젝트명>_예정노선_<연도>.zip` (이름, 바이트)."""
+def company_zip(project_name: str, label: str | int, files: dict[str, bytes]) -> tuple[str, bytes]:
+    """업체에 보낼 `<프로젝트명>_예정노선_<연도 | 연도_선_조각>.zip` (이름, 바이트)."""
     buffer = io.BytesIO()
     with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
         for name, data in files.items():
             archive.writestr(name, data)
-    return f"{safe_name(project_name)}_예정노선_{year}.zip", buffer.getvalue()
+    return f"{safe_name(project_name)}_예정노선_{safe_name(str(label))}.zip", buffer.getvalue()

D04_PlanConfirm/D04_PlanConfirm_Link.py

@@ -4,9 +4,9 @@
   new = 임도 프로젝트 새로(B02 저장소 함수) · existing = 고른 우리 회사 임도 프로젝트
 - `GET  /api/projects/{id}/plan-source` — 임도 프로젝트가 이어 받은 예정노선 · 업데이트 있음
 
-연결 = 그 연도 shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에 원래 이름으로 +
-`input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길. 입력이 다 차면
-B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림.
+연결 = 지정 행의 조각(옛 = 연도) shp 한 벌(D04 내보내기 함수)을 B03 입력 `input/shp/` 에
+원래 이름으로 + `input_files` 행 + metadata.json(B03 분석 함수) — 파일로 올린 것과 같은 길.
+입력이 다 차면 B03 완료 판정(옛 결과 지우기 확인 · 단계 되돌림) 뒤 WF1. 지형은 업체가 올림.
 연결 시점 복사 · 원본이 바뀌면 알림만(내용 지문 비교 · 자동 반영 없음).
 """
 
@@ -50,8 +50,7 @@ from common_util.common_util_auth import require_company
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_workflow_state import is_analysis_running, merge_stage_params
 from config.config_db import get_db_pool
-from D04_PlanConfirm.D04_PlanConfirm_Export import build_export
-from D04_PlanConfirm.D04_PlanConfirm_Router import year_routes_from_rows
+from D04_PlanConfirm.D04_PlanConfirm_Export import YearRoute, build_export
 
 router = APIRouter(prefix="/api/projects", tags=["D04 Plan Link"])
 dashboard_router = APIRouter(prefix="/api/dashboard", tags=["D04 Plan Link"])
@@ -102,18 +101,53 @@ async def _fetch_one(sql: str, args: tuple) -> dict[str, Any] | None:
         return await cursor.fetchone()
 
 
+# 지정 행의 노선 = 조각 행(조각 지정) · 연도 행(옛 연도 지정) — 확정은 연도 행
+PIECE_JOIN = (
+    " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year"
+    " LEFT JOIN plan_route_pieces c ON c.project_id = a.project_id AND c.piece_id = a.piece_id"
+)
+PIECE_COLUMNS = (
+    " a.piece_id, c.epsg AS piece_epsg, c.route_json AS piece_json,"
+    " c.from_m, c.to_m, c.length_m AS piece_length_m,"
+)
+
+
+def unit_row(row: dict[str, Any]) -> dict[str, Any]:
+    """조각 지정이면 `epsg` · `route_json` 을 조각 것으로 갈아 끼움(누가 = 선 위 거리 ·
+    길이 = 조각) — 아래 지문 · shp · 새 임도 길이가 그대로 돎. 조각 칸은 뺌."""
+    out = {k: v for k, v in row.items() if k not in ("piece_epsg", "piece_json", "from_m", "to_m")}
+    out.pop("piece_length_m", None)
+    if not row.get("piece_id"):
+        return out
+    body = _parsed(row.get("piece_json")) or {}
+    out["epsg"] = row.get("piece_epsg")
+    out["route_json"] = (
+        {
+            "planned": body["planned"],
+            "start_chainage_m": row["from_m"],
+            "end_chainage_m": row["to_m"],
+            "length_m": row["piece_length_m"],
+        }
+        if body.get("planned")
+        else None
+    )
+    return out
+
+
 async def load_assignee(assignee_id: int) -> dict[str, Any] | None:
-    """지정 행 + 예정노선 프로젝트 칸 + 그 연도 노선."""
-    return await _fetch_one(
+    """지정 행 + 예정노선 프로젝트 칸 + 그 조각(옛 = 연도) 노선."""
+    row = await _fetch_one(
         "SELECT a.id, a.project_id, a.year, a.company_id, a.linked_project_id,"
+        f"{PIECE_COLUMNS}"
         " p.name, p.region, p.road_type, p.design_speed_kph, p.route_start_m,"
         " y.epsg, y.route_json, y.confirmed_at"
         " FROM plan_route_assignees a"
         " JOIN projects p ON p.id = a.project_id AND p.deleted_at IS NULL"
-        " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year"
+        f"{PIECE_JOIN}"
         " WHERE a.id = %s",
         (assignee_id,),
     )
+    return unit_row(row) if row else None
 
 
 async def load_road_project(project_id: str) -> dict[str, Any] | None:
@@ -152,11 +186,17 @@ async def record_link(
 
 
 def year_files(assignee: dict[str, Any]) -> dict[str, bytes]:
-    """그 연도 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
-    row = {**assignee, "route_json": _parsed(assignee["route_json"])}
-    routes = year_routes_from_rows([row])
-    bundle = build_export(assignee["name"], assignee["region"] or "", routes, with_xlsx=False)
-    return bundle.by_year[int(assignee["year"])]
+    """그 조각(옛 = 연도) shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
+    body = _parsed(assignee["route_json"]) or {}
+    route = YearRoute(
+        year=int(assignee["year"]),
+        points=[(p[0], p[1]) for p in body["planned"]],
+        crs_epsg=int(assignee["epsg"]),
+        start_station_m=body.get("start_chainage_m"),
+        end_station_m=body.get("end_chainage_m"),
+    )
+    bundle = build_export(assignee["name"], assignee["region"] or "", [route], with_xlsx=False)
+    return bundle.by_year[route.key]
 
 
 async def _supersede_route_set(connection: aiomysql.Connection, project_id: str) -> None:
@@ -285,7 +325,7 @@ async def seed_design_speed(project_id: str, speed: Any) -> None:
 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="지정 업체 행을 찾을 수 없습니다.")
@@ -319,7 +359,13 @@ async def link_plan_route(
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     fingerprint = route_fingerprint(assignee["route_json"])
     await record_link(project_id, assignee, fingerprint, assignee["design_speed_kph"])
-    return {"status": "success", "project_id": project_id, "year": assignee["year"], **result}
+    return {
+        "status": "success",
+        "project_id": project_id,
+        "year": assignee["year"],
+        "piece_id": assignee.get("piece_id"),
+        **result,
+    }
 
 
 @router.get("/{project_id}/plan-source", response_model=None)
@@ -327,22 +373,25 @@ async def plan_source(project_id: UUID) -> dict[str, Any]:
     """임도 프로젝트가 이어 받은 예정노선(B03 머리 띠) — 없으면 linked False."""
     row = await _fetch_one(
         "SELECT p.source_fingerprint, s.name AS plan_name, a.id AS assignee_id, a.year,"
-        " y.route_json, y.confirmed_at"
+        f"{PIECE_COLUMNS}"
+        " y.epsg, y.route_json, y.confirmed_at"
         " FROM projects p"
         " JOIN projects s ON s.id = p.source_project_id AND s.project_kind = 'PLAN'"
         " JOIN plan_route_assignees a ON a.linked_project_id = p.id"
-        " LEFT JOIN plan_route_years y ON y.project_id = a.project_id AND y.year = a.year"
+        f"{PIECE_JOIN}"
         " WHERE p.id = %s LIMIT 1",
         (str(project_id),),
     )
     if not row:
         return {"status": "success", "linked": False}
+    row = unit_row(row)
     return {
         "status": "success",
         "linked": True,
         "assignee_id": row["assignee_id"],
         "plan_name": row["plan_name"],
         "year": row["year"],
+        "piece_id": row.get("piece_id"),
         "update_available": update_available(
             row["route_json"], row["source_fingerprint"], row["confirmed_at"] is not None
         ),

D04_PlanConfirm/D04_PlanConfirm_Repository.py

@@ -41,13 +41,45 @@ async def list_year_routes(project_id: str) -> list[dict[str, Any]]:
     return rows
 
 
+async def list_pieces(project_id: str) -> list[dict[str, Any]]:
+    """조각 행(연도 · 선 차례 · 선 위 거리 차례) + `line_name`(본선 · 갈래 N = 정본 선 차례)."""
+    async with (
+        get_db_pool().acquire() as connection,
+        connection.cursor(aiomysql.DictCursor) as cursor,
+    ):
+        await cursor.execute(
+            "SELECT piece_id, line_id, year, from_m, to_m, length_m, epsg, route_json"
+            " FROM plan_route_pieces WHERE project_id = %s",
+            (project_id,),
+        )
+        rows = list(await cursor.fetchall())
+        await cursor.execute(
+            "SELECT network_json FROM plan_networks WHERE project_id = %s", (project_id,)
+        )
+        network = await cursor.fetchone()
+    lines = []
+    if network:
+        raw = network["network_json"]
+        body = json.loads(raw) if isinstance(raw, str | bytes) else raw
+        lines = [line["id"] for line in (body or {}).get("lines", [])]
+    order = {line: i for i, line in enumerate(lines)}
+    for row in rows:
+        raw = row["route_json"]
+        row["route_json"] = json.loads(raw) if isinstance(raw, str | bytes) else raw
+        index = order.get(row["line_id"], len(order))
+        row["line_name"] = "본선" if index == 0 else f"갈래 {index}"
+    rows.sort(key=lambda r: (r["year"], order.get(r["line_id"], len(order)), r["from_m"]))
+    return rows
+
+
 async def list_assignees(project_id: str) -> list[dict[str, Any]]:
     async with (
         get_db_pool().acquire() as connection,
         connection.cursor(aiomysql.DictCursor) as cursor,
     ):
         await cursor.execute(
-            "SELECT id, year, company_id, company_name, email, notified_at, linked_project_id"
+            "SELECT id, year, piece_id, company_id, company_name, email, notified_at,"
+            " linked_project_id"
             " FROM plan_route_assignees WHERE project_id = %s ORDER BY year, id",
             (project_id,),
         )
@@ -89,8 +121,16 @@ async def apply_assignee_changes(
             for row in inserts:
                 await cursor.execute(
                     "INSERT INTO plan_route_assignees"
-                    " (project_id, year, company_id, company_name, email) VALUES (%s,%s,%s,%s,%s)",
-                    (project_id, row["year"], row["company_id"], row["company_name"], row["email"]),
+                    " (project_id, year, piece_id, company_id, company_name, email)"
+                    " VALUES (%s,%s,%s,%s,%s,%s)",
+                    (
+                        project_id,
+                        row["year"],
+                        row.get("piece_id"),
+                        row["company_id"],
+                        row["company_name"],
+                        row["email"],
+                    ),
                 )
             await connection.commit()
         except Exception:

D04_PlanConfirm/D04_PlanConfirm_Router.py

@@ -1,10 +1,14 @@
-"""D 예정노선 확정 · 업체 지정 API (요구 345 · PLAN 531-1 · 531-2).
+"""D 예정노선 확정 · 업체 지정 API (요구 345 · 391 · PLAN 531-1 · 531-2 · 563-2).
 
-- `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(산림청 = 전 조각 + 전체 · 공유 = 자기 조각)
+
+단위 = 조각(`plan_route_pieces` · 구역 × 갈래 · 요구 391) · 조각이 없는 연도 행(옛 프로젝트)은
+연도 하나가 단위. 확정은 연도 단위(`plan_route_years.confirmed_at`). 업체 행 `piece_id` 가 비면
+옛 연도 지정 — 그 연도 첫 단위에 붙음.
 
 지정 · 확정은 산림청(시스템 관리자 또는 메일 *.go.kr)만. 지정받은 회사는 공용 접근 판정이
 GET 만 통과시킴. 계산 자리 ③ 서버 단독 — 내보내기 파일은 저장된 `route_json` 으로만 만듦.
@@ -34,6 +38,7 @@ from D04_PlanConfirm.D04_PlanConfirm_Export import (
     build_export,
     company_zip,
     safe_name,
+    station_label,
 )
 
 logger = logging.getLogger(__name__)
@@ -42,6 +47,7 @@ router = APIRouter(prefix="/api/projects", tags=["D04 Plan Confirm"])
 
 class AssigneeIn(BaseModel):
     year: int
+    piece_id: str | None = None  # 조각 — 비면 옛 연도 지정(조각 없는 프로젝트)
     company_id: int | None = None  # 등록 회사 — 없으면 수기 업체
     company_name: str | None = None
     email: str | None = None
@@ -65,10 +71,11 @@ def _require_forest(session: dict[str, Any]) -> None:
 
 
 def _key(row: dict[str, Any]) -> tuple:
-    """같은 업체 판정 — 등록 회사는 id · 수기 업체는 메일(대소문자 무시)."""
+    """같은 업체 판정 — 같은 조각(옛 = 연도)에서 등록 회사 = id · 수기 = 메일(대소문자 무시)."""
+    unit = (row["year"], row.get("piece_id"))
     if row.get("company_id") is not None:
-        return (row["year"], "c", row["company_id"])
-    return (row["year"], "e", (row.get("email") or "").strip().lower())
+        return (*unit, "c", row["company_id"])
+    return (*unit, "e", (row.get("email") or "").strip().lower())
 
 
 def plan_assignee_changes(
@@ -90,6 +97,7 @@ async def _resolve_assignee(item: AssigneeIn) -> dict[str, Any]:
     """등록 회사면 이름 · 대표 메일을 DB 에서, 수기면 이름 + 메일을 검사."""
     email = (item.email or "").strip()
     name = (item.company_name or "").strip()
+    unit = {"year": item.year, "piece_id": item.piece_id}
     if item.company_id is not None:
         company = await repo.get_company_contact(item.company_id)
         if not company:
@@ -97,14 +105,14 @@ async def _resolve_assignee(item: AssigneeIn) -> dict[str, Any]:
                 status_code=400, detail=f"등록 회사를 찾을 수 없습니다: {item.company_id}"
             )
         return {
-            "year": item.year,
+            **unit,
             "company_id": item.company_id,
             "company_name": company["name"],
             "email": email or company.get("email") or None,
         }
     if not name or "@" not in email:
         raise HTTPException(status_code=400, detail="업체명과 메일 주소가 필요합니다.")
-    return {"year": item.year, "company_id": None, "company_name": name, "email": email}
+    return {**unit, "company_id": None, "company_name": name, "email": email}
 
 
 async def _plan_project(project_id: UUID) -> dict[str, Any]:
@@ -116,16 +124,89 @@ async def _plan_project(project_id: UUID) -> dict[str, Any]:
     return project
 
 
+def route_units(rows: list[dict[str, Any]], pieces: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """표 한 줄 단위(연도 차례) — 조각마다 · 조각이 없는 연도 행(옛 · 분석 전)은 연도 하나."""
+    confirmed = {int(r["year"]): r.get("confirmed_at") for r in rows}
+    units = [
+        {
+            "piece_id": p["piece_id"],
+            "line": p["line_id"],
+            "line_name": p.get("line_name"),
+            "year": int(p["year"]),
+            "epsg": int(p["epsg"]),
+            "planned": (p.get("route_json") or {}).get("planned") or [],
+            "start_chainage_m": float(p["from_m"]),
+            "end_chainage_m": float(p["to_m"]),
+            "length_m": float(p["length_m"]),
+            "confirmed_at": confirmed.get(int(p["year"])),
+        }
+        for p in pieces
+    ]
+    taken = {u["year"] for u in units}
+    for row in rows:
+        if int(row["year"]) in taken:
+            continue
+        body = row.get("route_json") or {}
+        units.append(
+            {
+                "piece_id": None,
+                "line": None,
+                "line_name": None,
+                "year": int(row["year"]),
+                "epsg": row.get("epsg"),
+                "planned": body.get("planned") or [],
+                "start_chainage_m": body.get("start_chainage_m"),
+                "end_chainage_m": body.get("end_chainage_m"),
+                "length_m": body.get("length_m"),
+                "confirmed_at": row.get("confirmed_at"),
+            }
+        )
+    units.sort(key=lambda u: u["year"])  # 같은 연도 안 = 조각 차례 그대로
+    return units
+
+
+def unit_key(unit: dict[str, Any]) -> str | int:
+    """내보내기 묶음 열쇠 — 조각 id · 옛 연도."""
+    return unit["piece_id"] or unit["year"]
+
+
+def unit_label(unit: dict[str, Any]) -> str:
+    """zip 폴더 · 첨부 이름 — 「2026」(옛) · 「2026_갈래1_p3」(조각)."""
+    if not unit["piece_id"]:
+        return str(unit["year"])
+    line = (unit["line_name"] or unit["line"]).replace(" ", "")
+    return f"{unit['year']}_{line}_{unit['piece_id']}"
+
+
+def unit_of(assignee: dict[str, Any], units: list[dict[str, Any]]) -> dict[str, Any] | None:
+    """업체 행의 단위 — 조각 지정 = 그 조각 · 옛 연도 지정 = 그 연도 첫 단위."""
+    piece = assignee.get("piece_id")
+    for unit in units:
+        if (unit["piece_id"] == piece) if piece else (unit["year"] == assignee["year"]):
+            return unit
+    return None
+
+
+def _saved(unit: dict[str, Any]) -> bool:
+    return len(unit["planned"]) >= 2
+
+
 def visible_to(
-    session: dict[str, Any], years: list[dict[str, Any]], assignees: list[dict[str, Any]]
+    session: dict[str, Any], units: list[dict[str, Any]], assignees: list[dict[str, Any]]
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
-    """산림청 = 전부 · 지정 회사 = 자기 회사가 지정된 연도 줄 · 자기 업체 행만(남의 이름 · 메일 뺌)."""
+    """산림청 = 전부 · 지정 회사 = 자기 회사가 지정된 단위 · 자기 업체 행만(남의 이름 · 메일 뺌)."""
     if is_forest_service(session):
-        return years, assignees
+        return units, assignees
     company = session.get("company_id")
     mine = [a for a in assignees if a["company_id"] is not None and a["company_id"] == company]
-    own_years = {a["year"] for a in mine}
-    return [y for y in years if y["year"] in own_years], mine
+    own = [unit_of(a, units) for a in mine]
+    return [u for u in units if any(u is o for o in own)], mine
+
+
+async def _units(project_id: UUID) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """(연도 행, 단위)."""
+    rows = await repo.list_year_routes(str(project_id))
+    return rows, route_units(rows, await repo.list_pieces(str(project_id)))
 
 
 @router.get("/{project_id}/plan/assignees", response_model=None)
@@ -141,10 +222,19 @@ async def get_plan_assignees(
 async def put_plan_assignees(
     project_id: UUID, request: AssigneesRequest, session: dict = Depends(verify_session)
 ) -> dict[str, Any]:
-    """보낸 연도의 지정 업체를 그대로 맞춤 — 연도를 안 보낸 곳은 건드리지 않음."""
+    """보낸 연도의 지정 업체를 그대로 맞춤 — 연도를 안 보낸 곳은 건드리지 않음.
+
+    조각 지정이면 연도는 조각 행에서 읽음(보낸 연도는 무시)."""
     _require_forest(session)
     await _plan_project(project_id)
     known = {row["year"] for row in await repo.list_year_routes(str(project_id))}
+    piece_years = {p["piece_id"]: int(p["year"]) for p in await repo.list_pieces(str(project_id))}
+    for item in request.assignees:
+        if item.piece_id is None:
+            continue
+        if item.piece_id not in piece_years:
+            raise HTTPException(status_code=400, detail=f"조각이 없습니다: {item.piece_id}")
+        item.year = piece_years[item.piece_id]
     years = {a.year for a in request.assignees} | set(request.years or [])
     unknown = sorted(years - known)
     if unknown:
@@ -156,28 +246,27 @@ async def put_plan_assignees(
     return {"status": "success", "assignees": await repo.list_assignees(str(project_id))}
 
 
-def year_routes_from_rows(rows: list[dict[str, Any]]) -> list[YearRoute]:
-    """분석이 저장된 연도 → 내보내기 입력 (D03 `route_json["planned"]` = 프로젝트 좌표 꺾은선)."""
-    routes = []
-    for row in rows:
-        body = row.get("route_json")
-        if not body or not body.get("planned"):
-            continue
-        routes.append(
-            YearRoute(
-                year=int(row["year"]),
-                points=[(p[0], p[1]) for p in body["planned"]],
-                crs_epsg=int(row["epsg"]),
-                start_station_m=body.get("start_chainage_m"),
-                end_station_m=body.get("end_chainage_m"),
-            )
+def unit_routes(units: list[dict[str, Any]]) -> list[YearRoute]:
+    """저장된 단위 → 내보내기 입력(`planned` = 프로젝트 좌표 꺾은선)."""
+    return [
+        YearRoute(
+            year=unit["year"],
+            points=[(p[0], p[1]) for p in unit["planned"]],
+            crs_epsg=int(unit["epsg"]),
+            start_station_m=unit["start_chainage_m"],
+            end_station_m=unit["end_chainage_m"],
+            piece=unit["piece_id"],
+            line=(unit["line_name"] or unit["line"] or "").replace(" ", "") or None,
         )
-    return routes
+        for unit in units
+        if _saved(unit)
+    ]
 
 
 def _result(assignee: dict[str, Any], state: str) -> dict[str, Any]:
     return {
         "year": assignee["year"],
+        "piece_id": assignee.get("piece_id"),
         "company_name": assignee["company_name"],
         "email": assignee["email"],
         "state": state,
@@ -188,10 +277,11 @@ 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)
-    routes = year_routes_from_rows(await repo.list_year_routes(str(project_id)))
+    _, units = await _units(project_id)
+    routes = unit_routes(units)
     saved = {r.year for r in routes}
     years = sorted(set(request.years) if request.years else saved)
     missing = [y for y in years if y not in saved]
@@ -201,7 +291,7 @@ async def finalize_plan(
         )
     await repo.mark_confirmed(str(project_id), years)
 
-    # 누가거리가 연도마다 이어지므로 저장된 전 연도로 한 번에 만들고, 보낼 때만 연도를 고름.
+    # 옛 누가거리가 연도마다 이어지므로 저장된 전 단위로 한 번에 만들고, 보낼 때만 고름.
     first = float(project.get("route_start_m") or 0.0)
     bundle = await asyncio.to_thread(
         build_export, project["name"], project["region"] or "", routes, first
@@ -217,12 +307,17 @@ async def finalize_plan(
         if not assignee["email"]:
             results.append(_result(assignee, "no_email"))
             continue
-        zip_name, zip_bytes = company_zip(project["name"], year, bundle.by_year[year])
+        unit = unit_of(assignee, units)
+        if unit is None or unit_key(unit) not in bundle.by_year:
+            results.append(_result(assignee, "no_route"))
+            continue
+        key = unit_key(unit)
+        zip_name, zip_bytes = company_zip(project["name"], unit_label(unit), bundle.by_year[key])
         subject, body = build_completion_mail(
             project_name=project["name"],
             region=project["region"],
             year=year,
-            length_m=bundle.lengths_m[year],
+            length_m=bundle.lengths_m[key],
             recipient=assignee["company_name"] or assignee["email"],
             attachment_names=[zip_name],
         )
@@ -263,25 +358,58 @@ def year_summary(row: dict[str, Any]) -> dict[str, Any]:
     return out
 
 
+def unit_summary(unit: dict[str, Any]) -> dict[str, Any]:
+    """표 한 줄(조각 · 옛 연도) — 저장 전이면 saved False · label 은 조각만(「본선 0+0~12+10」)."""
+    out = {key: unit[key] for key in ("piece_id", "line", "line_name", "year", "length_m")}
+    out.update(saved=_saved(unit), confirmed_at=None, label=None)
+    if unit["confirmed_at"]:
+        out["confirmed_at"] = str(unit["confirmed_at"])
+    if out["saved"]:
+        epsg = int(unit["epsg"])
+        out.update(
+            start_chainage_m=unit["start_chainage_m"],
+            end_chainage_m=unit["end_chainage_m"],
+            start=_end_point(unit["planned"][0], epsg),
+            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"]
+        )
+    return out
+
+
+def piece_label(line_name: str, from_m: float, to_m: float) -> str:
+    """조각 이름 「본선 0+0~12+10」(측점 = 선마다 No.0)."""
+    return f"{line_name} {station_label(from_m)}~{station_label(to_m)}"
+
+
 @router.get("/{project_id}/plan/summary", response_model=None)
 async def get_plan_summary(
     project_id: UUID, session: dict = Depends(verify_session)
 ) -> dict[str, Any]:
+    """`years` = 연도 행(확정 고르기) · `pieces` = 표 한 줄 단위(조각 · 옛 연도)."""
     project = await _plan_project(project_id)
-    rows = await repo.list_year_routes(str(project_id))
-    years, assignees = visible_to(
-        session, [year_summary(row) for row in rows], await repo.list_assignees(str(project_id))
-    )
-    return {"status": "success", "name": project["name"], "years": years, "assignees": assignees}
+    rows, units = await _units(project_id)
+    units, assignees = visible_to(session, units, await repo.list_assignees(str(project_id)))
+    own_years = {u["year"] for u in units}
+    years = [year_summary(row) for row in rows if int(row["year"]) in own_years]
+    return {
+        "status": "success",
+        "name": project["name"],
+        "years": years,
+        "pieces": [unit_summary(u) for u in units],
+        "assignees": assignees,
+    }
 
 
-def export_zip(bundle: Any, years: list[int], with_all: bool) -> bytes:
-    """연도마다 `<연도>/` · 전체는 `전체/` — 좌표표 이름이 연도마다 같아 폴더로 가름."""
+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 year in years:
-            for name, data in bundle.by_year[year].items():
-                archive.writestr(f"{year}/{name}", data)
+        for unit in units:
+            for name, data in bundle.by_year[unit_key(unit)].items():
+                archive.writestr(f"{unit_label(unit)}/{name}", data)
         if with_all:
             for name, data in bundle.all.items():
                 archive.writestr(f"전체/{name}", data)
@@ -290,30 +418,24 @@ def export_zip(bundle: Any, years: list[int], with_all: bool) -> bytes:
 
 @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)
-    rows = await repo.list_year_routes(str(project_id))
-    routes = year_routes_from_rows(rows)
+    _, units = await _units(project_id)
+    routes = unit_routes(units)
     if not routes:
         raise HTTPException(status_code=409, detail="분석 [저장]이 된 연도가 없습니다.")
     forest = is_forest_service(session)
-    if forest:
-        years = [r.year for r in routes]
-    else:
-        confirmed = {int(r["year"]) for r in rows if r.get("confirmed_at")}
-        mine = {
-            a["year"]
-            for a in await repo.list_assignees(str(project_id))
-            if a["company_id"] is not None and a["company_id"] == session.get("company_id")
-        }
-        years = sorted({r.year for r in routes} & confirmed & mine)
-        if not years:
-            raise HTTPException(status_code=403, detail="받을 수 있는 연도가 없습니다.")
+    picked = [u for u in units if _saved(u)]
+    if not forest:
+        own, _ = visible_to(session, picked, await repo.list_assignees(str(project_id)))
+        picked = [u for u in own if u["confirmed_at"]]
+        if not picked:
+            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
     )
-    data = await asyncio.to_thread(export_zip, bundle, years, forest)
+    data = await asyncio.to_thread(export_zip, bundle, picked, forest)
     name = f"{safe_name(project['name'])}_예정노선.zip"
     return Response(
         content=data,

resources/tester/test_531_3_plan_link.py

@@ -438,9 +438,13 @@ def plan_repo(monkeypatch):
             {"id": 10, "year": 2028, "company_id": 6, "company_name": "다", "email": "c@da.kr"},
         ]
 
+    async def no_pieces(_pid):  # 옛 프로젝트 — 조각 없음
+        return []
+
     monkeypatch.setattr(repo, "get_plan_project", project)
     monkeypatch.setattr(repo, "list_year_routes", years)
     monkeypatch.setattr(repo, "list_assignees", assignees)
+    monkeypatch.setattr(repo, "list_pieces", no_pieces)
 
 
 def _zip_names(response):

resources/tester/test_531_d04_export_finalize.py

@@ -328,9 +328,13 @@ def fake_db(monkeypatch):
         state["mails"].append((to_email, subject, attachments))
         return True
 
+    async def no_pieces(_pid):  # 옛 프로젝트 — 조각 없음
+        return []
+
     monkeypatch.setattr(repo, "get_plan_project", project)
     monkeypatch.setattr(repo, "list_year_routes", years)
     monkeypatch.setattr(repo, "list_assignees", assignees)
+    monkeypatch.setattr(repo, "list_pieces", no_pieces)
     monkeypatch.setattr(repo, "mark_confirmed", confirmed)
     monkeypatch.setattr(repo, "mark_notified", notified)
     monkeypatch.setattr(repo, "apply_assignee_changes", apply)

resources/tester/test_563_2_plan_pieces.py

@@ -0,0 +1,331 @@
+# -*- coding: utf-8 -*-
+"""563-2 예정노선 아래 단계 = 조각 단위(요구 391).
+
+견본 = 같은 연도 조각 둘(본선 p1 · 갈래 p3, 2026) + 본선 p2(2027) · 업체 둘(p1 = 회사 5 · p3 = 회사 6).
+- D04 요약 = 조각 줄 · 이름 「본선 0+0~12+10」 · 지정 회사 = 자기 조각 · 자기 행만
+- 내보내기 = 조각마다 폴더 · 전체 = 선마다 레코드 · 지정 회사 = 확정된 자기 조각만
+- 확정 메일 = 업체마다 자기 조각 zip(남의 조각 안 샘)
+- 업체 저장 = 조각 id 검사 · 연도는 조각에서 · 같은 업체 두 조각 = 두 행
+- 연결 · 가져오기 = 조각 꺾은선 shp · 받은 목록 = 조각 이름 · 길이
+"""
+
+from __future__ import annotations
+
+import asyncio
+import io
+import json
+import sys
+import tempfile
+import zipfile
+from pathlib import Path
+from uuid import uuid4
+
+import geopandas as gpd
+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 B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata  # noqa: E402
+from D01_PlanRegister import D01_PlanRegister_Router as d01  # 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())
+P1 = [[210000.0, 380000.0], [210200.0, 380150.0]]  # 본선 0 ~ 250
+P2 = [[210200.0, 380150.0], [210300.0, 380400.0]]  # 본선 250 ~ 519.258
+P3 = [[210200.0, 380150.0], [210330.0, 380150.0]]  # 갈래 0 ~ 130
+L2 = (100.0**2 + 250.0**2) ** 0.5
+
+FOREST = {"role": "SYSTEM_ADMIN", "email": "root@x.kr", "company_id": 1, "user_id": 1}
+FIRM5 = {"role": "USER", "email": "a@ga.kr", "company_id": 5, "user_id": 3}
+FIRM6 = {"role": "USER", "email": "b@na.kr", "company_id": 6, "user_id": 4}
+
+
+def _piece(pid, line, name, year, start, end, planned):
+    return {
+        "piece_id": pid,
+        "line_id": line,
+        "line_name": name,
+        "year": year,
+        "from_m": start,
+        "to_m": end,
+        "length_m": end - start,
+        "epsg": 5186,
+        "route_json": {"planned": planned, "stations": [start, end]},
+    }
+
+
+PIECES = [
+    _piece("p1", "main", "본선", 2026, 0.0, 250.0, P1),
+    _piece("p3", "b1", "갈래 1", 2026, 0.0, 130.0, P3),
+    _piece("p2", "main", "본선", 2027, 250.0, 250.0 + L2, P2),
+]
+
+
+def _assignee(aid, piece, year, company, name, email, notified=None):
+    return {
+        "id": aid,
+        "year": year,
+        "piece_id": piece,
+        "company_id": company,
+        "company_name": name,
+        "email": email,
+        "notified_at": notified,
+        "linked_project_id": None,
+    }
+
+
+@pytest.fixture
+def plan(monkeypatch):
+    state = {
+        "years": [
+            {"year": 2026, "epsg": 5186, "route_json": {"planned": P1}, "confirmed_at": "x"},
+            {"year": 2027, "epsg": 5186, "route_json": {"planned": P2}, "confirmed_at": None},
+        ],
+        "assignees": [
+            _assignee(7, "p1", 2026, 5, "가", "a@ga.kr"),
+            _assignee(8, "p3", 2026, 6, "나", "b@na.kr"),
+        ],
+        "mails": [],
+        "applied": None,
+    }
+
+    async def project(_pid):
+        return {"id": PLAN_ID, "name": "시험", "region": "일월.용화", "project_kind": "PLAN"}
+
+    async def years(_pid):
+        return state["years"]
+
+    async def pieces(_pid):
+        return PIECES
+
+    async def assignees(_pid):
+        return state["assignees"]
+
+    async def noop(*a):
+        return None
+
+    async def apply(_pid, d, u, i):
+        state["applied"] = (d, u, i)
+
+    async def contact(cid):
+        return {"id": cid, "name": f"등록{cid}", "email": f"m{cid}@reg.kr"}
+
+    async def send(*, to_email, subject, body, attachments):
+        state["mails"].append((to_email, attachments))
+        return True
+
+    monkeypatch.setattr(repo, "get_plan_project", project)
+    monkeypatch.setattr(repo, "list_year_routes", years)
+    monkeypatch.setattr(repo, "list_pieces", pieces)
+    monkeypatch.setattr(repo, "list_assignees", assignees)
+    monkeypatch.setattr(repo, "mark_confirmed", noop)
+    monkeypatch.setattr(repo, "mark_notified", noop)
+    monkeypatch.setattr(repo, "apply_assignee_changes", apply)
+    monkeypatch.setattr(repo, "get_company_contact", contact)
+    monkeypatch.setattr(d04, "send_completion_mail", send)
+    return state
+
+
+def _names(response) -> list[str]:
+    return sorted(zipfile.ZipFile(io.BytesIO(response.body)).namelist())
+
+
+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"),
+    ]
+    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]
+    # 같은 연도 다른 업체 — 자기 조각 · 자기 행만
+    for session, piece, aid in ((FIRM5, "p1", 7), (FIRM6, "p3", 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]
+        assert [y["year"] for y in mine["years"]] == [2026]
+    assert "b@na.kr" not in str(asyncio.run(d04.get_plan_summary(PLAN_ID, FIRM5)))
+
+
+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",
+        "전체",
+    }
+    assert sum(n.endswith(".shp") for n in names) == 4
+    with zipfile.ZipFile(io.BytesIO(asyncio.run(d04.export_plan(PLAN_ID, FOREST)).body)) as z:
+        with tempfile.TemporaryDirectory() as tmp:
+            for n in z.namelist():
+                if n.startswith("전체/") and not n.endswith(".xlsx"):
+                    (Path(tmp) / n.split("/")[1]).write_bytes(z.read(n))
+            whole = gpd.read_file(next(Path(tmp).glob("*.shp")))
+    # 전체 = 선마다 한 레코드 — 본선(p1 + p2 이음) · 갈래
+    assert len(whole) == 2
+    assert sorted(round(g.length, 1) for g in whole.to_crs(5186).geometry) == [
+        130.0,
+        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"}
+    plan["years"][0]["confirmed_at"] = None  # 확정 전 = 받을 것 없음
+    with pytest.raises(HTTPException) as err:
+        asyncio.run(d04.export_plan(PLAN_ID, FIRM6))
+    assert err.value.status_code == 403
+
+
+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"),
+    ]
+    for (to, attachments), piece in zip(plan["mails"], ("본선_p1", "갈래1_p3"), strict=True):
+        name, blob, _ = attachments[0]
+        assert name == f"시험_예정노선_2026_{piece}.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")
+
+
+def test_put_assignees_by_piece(plan):
+    def put(items, years=None):
+        request = d04.AssigneesRequest(assignees=items, years=years)
+        return asyncio.run(d04.put_plan_assignees(PLAN_ID, request, FOREST))
+
+    with pytest.raises(HTTPException) as err:
+        put([d04.AssigneeIn(year=2026, piece_id="없음", company_id=5)])
+    assert err.value.status_code == 400
+    # 같은 업체(회사 5)를 같은 연도 두 조각에 — 두 행 · 연도는 조각에서(보낸 2099 무시)
+    put(
+        [
+            d04.AssigneeIn(year=2026, piece_id="p1", company_id=5),
+            d04.AssigneeIn(year=2099, piece_id="p3", company_id=5),
+        ],
+        years=[2026],
+    )
+    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)]
+
+
+def test_legacy_year_assignee_sits_on_first_piece(plan):
+    """조각이 생기기 전 연도 지정(piece_id 없음) — 그 연도 첫 조각에 붙어 보임."""
+    plan["assignees"] = [_assignee(9, None, 2026, 5, "가", "a@ga.kr")]
+    mine = asyncio.run(d04.get_plan_summary(PLAN_ID, FIRM5))
+    assert [r["piece_id"] for r in mine["pieces"]] == ["p1"]
+
+
+def test_link_uses_piece_route():
+    row = {
+        "id": 7,
+        "year": 2026,
+        "piece_id": "p3",
+        "piece_epsg": 5186,
+        "piece_json": json.dumps({"planned": P3, "stations": [0.0, 130.0]}),
+        "from_m": 0.0,
+        "to_m": 130.0,
+        "piece_length_m": 130.0,
+        "epsg": 5186,
+        "route_json": json.dumps({"planned": P1, "length_m": 380.0}),
+        "name": "2026 홍천 노선 계획",
+        "region": "일월.용화",
+    }
+    unit = link.unit_row(row)
+    assert unit["route_json"]["planned"] == P3 and unit["route_json"]["length_m"] == 130.0
+    assert "piece_json" not in unit and "from_m" not in unit
+    assert link.route_fingerprint(unit["route_json"]) != link.route_fingerprint(row["route_json"])
+    files = link.year_files(unit)
+    with tempfile.TemporaryDirectory() as tmp:
+        for name, data in files.items():
+            (Path(tmp) / name).write_bytes(data)
+        meta = analyze_input_metadata(next(Path(tmp).glob("*.shp")))
+    assert meta["point_count"] == 2 and float(meta["attributes"]["종점누가"]) == 130.0
+    # 옛 연도 지정 = 연도 노선 그대로
+    legacy = link.unit_row({**row, "piece_id": None})
+    assert legacy["route_json"] == row["route_json"] and legacy["epsg"] == 5186
+
+
+class _Cursor:
+    def __init__(self, rows):
+        self.rows = rows
+
+    async def execute(self, sql, args=None):
+        self.sql = sql
+
+    async def fetchall(self):
+        return self.rows
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+
+class _Conn:
+    def __init__(self, rows):
+        self.rows = rows
+
+    def acquire(self):
+        return self
+
+    def cursor(self, *a):
+        return _Cursor(self.rows)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+
+def test_assigned_list_piece_label_length(monkeypatch):
+    rows = [
+        {
+            "assignee_id": 8,
+            "project_id": PLAN_ID,
+            "project_name": "시험",
+            "year": 2026,
+            "piece_id": "p3",
+            "piece_length_m": 130.0,
+            "linked_project_id": None,
+            "linked_project_name": None,
+            "source_fingerprint": None,
+            "route_json": json.dumps({"planned": P3}),
+            "confirmed_at": "x",
+        },
+        {
+            "assignee_id": 9,
+            "project_id": PLAN_ID,
+            "project_name": "옛",
+            "year": 2025,
+            "piece_id": None,
+            "piece_length_m": None,
+            "linked_project_id": None,
+            "linked_project_name": None,
+            "source_fingerprint": None,
+            "route_json": json.dumps({"planned": P1, "length_m": 250.0}),
+            "confirmed_at": "x",
+        },
+    ]
+
+    async def pieces(_pid):
+        return PIECES
+
+    monkeypatch.setattr(d01, "get_db_pool", lambda: _Conn(rows))
+    monkeypatch.setattr(repo, "list_pieces", pieces)
+    out = asyncio.run(d01.assigned_plan_routes(FIRM6))["routes"]
+    assert [(r["piece_id"], r["piece_label"], r["length_m"]) for r in out] == [
+        ("p3", "갈래 1 0+0~6+10", 130.0),
+        (None, None, 250.0),
+    ]