feat(D02·D03): 기준지정 시점 · 종점 목록 서버 꼴 — plan/points {starts, ends} · 옛 한 점 꼴 감쌈 · 저장된 좌표계 그대로 (PLAN 579-1 · 요구 440)

- D02 정본 칸 plan_base_json = {epsg, starts: [{x, y}], ends: [...]} · 옛 start · end 는 읽을 때 목록으로 감쌈(옮김 없음)
- 목록마다 상한 20 · 저장된 epsg 가 있으면 그대로(점이 늘어 평균 경도가 띠를 넘어도 D03 노선과 안 어긋남)
- 부르는 곳 없던 연도 잇기 코드(chain_years · with_tm · changed · _store) 걷음 — 시험도 지움
- D03 지형 입구 = 시점 · 종점 모두 · 정본 굳힘 = 새 꼴

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

D02_PlanBase/D02_PlanBase_Router.py

@@ -1,12 +1,14 @@
-"""D 예정노선 1단계 기준 지정 — 전체 시점 · 종점 두 점 정본 API (요구서 389 3장 1 · PLAN 560-2).
+"""D 예정노선 1단계 기준 지정 — 전체 시점 · 종점 목록 정본 API
+(요구서 389 3장 1 · 440 1 · PLAN 560-2 · 579-1).
 
-- `GET /api/projects/{id}/plan/points` — {epsg, start, end} · 점 = {lon, lat, x, y} · 빈 = null
-- `PUT …/plan/points` — {start: {lon, lat}, end: {lon, lat}} → TM 셈 → `projects.plan_base_json`
-  (`{"epsg", "start": {"x", "y"}, "end": {"x", "y"}}` · 좌표계 = 두 점 평균 경도의 TM 띠)
+- `GET /api/projects/{id}/plan/points` — {epsg, starts: [점], ends: [점]} · 점 = {lon, lat, x, y}
+- `PUT …/plan/points` — {starts: [{lon, lat}], ends: [...]} → TM 셈 → `projects.plan_base_json`
+  (`{"epsg", "starts": [{"x", "y"}], "ends": [...]}` · 좌표계 = 저장된 값 ·
+  없으면 점 평균 경도의 TM 띠)
+- 시점 · 종점은 짝 없는 따로 목록 · 옛 꼴(`start` · `end` 한 점씩)은 읽을 때 목록으로 감쌈
+  (옮김 없음).
 - 칸이 비면(옛 프로젝트) 첫 연도 시점 + 마지막 연도 종점으로 접어 보임 · 연도 행은 D03 [저장] 몫.
 - 고침 = 산림청만(지정 회사 GET 은 공용 판정 `require_project_access` 몫).
-
-연도 잇기 꼴(`chain_years` · `with_tm` · `changed` · `_store`)은 D03 [저장]이 씀.
 """
 
 from __future__ import annotations
@@ -25,7 +27,7 @@ from config.config_db import get_db_pool
 
 router = APIRouter(prefix="/api/projects", tags=["D02 Plan Base"])
 
-SAME_M = 0.001  # 1 mm 안이면 같은 지점 — 되저장으로 분석 노선을 지우지 않음
+MAX_POINTS = 20  # 시점 · 종점 목록마다 상한
 
 
 class LonLat(BaseModel):
@@ -33,60 +35,9 @@ class LonLat(BaseModel):
     lat: float = Field(..., ge=30, le=45)
 
 
-class YearPoints(BaseModel):
-    year: int
-    start: LonLat | None = None
-    end: LonLat | None = None
-
-
 class BaseRequest(BaseModel):
-    start: LonLat | None = None
-    end: LonLat | None = None
-
-
-def chain_years(year_start: int, year_end: int, items: list[YearPoints]) -> list[dict[str, Any]]:
-    """연도 범위를 잇기 — [{year, start, end}] 경위도 · 시점 = 직전 종점(첫 연도만 보낸 값)."""
-    by_year: dict[int, YearPoints] = {}
-    for item in items:
-        if not year_start <= item.year <= year_end:
-            raise HTTPException(
-                status_code=400,
-                detail=f"{item.year}년은 사업연도({year_start}~{year_end}) 밖입니다.",
-            )
-        if item.year in by_year:
-            raise HTTPException(status_code=400, detail=f"{item.year}년이 두 번 왔습니다.")
-        by_year[item.year] = item
-    rows: list[dict[str, Any]] = []
-    previous: LonLat | None = None
-    for year in range(year_start, year_end + 1):
-        item = by_year.get(year)
-        start = previous or (item.start if item else None)
-        end = item.end if item else None
-        if start is None or end is None:
-            side = "시점" if start is None else "종점"
-            raise HTTPException(status_code=400, detail=f"{year}년 {side}이 비었습니다.")
-        rows.append({"year": year, "start": start, "end": end})
-        previous = end
-    return rows
-
-
-def with_tm(rows: list[dict[str, Any]]) -> tuple[int, list[dict[str, Any]]]:
-    """(epsg, 행마다 start_x · start_y · end_x · end_y 더함)."""
-    lons = [p.lon for row in rows for p in (row["start"], row["end"])]
-    epsg = tm_epsg_for_lon(sum(lons) / len(lons))
-    out = []
-    for row in rows:
-        sx, sy = lonlat_to_tm(row["start"].lon, row["start"].lat, epsg)
-        ex, ey = lonlat_to_tm(row["end"].lon, row["end"].lat, epsg)
-        out.append({"year": row["year"], "start_x": sx, "start_y": sy, "end_x": ex, "end_y": ey})
-    return epsg, out
-
-
-def changed(new: dict[str, Any], old: dict[str, Any] | None, epsg: int) -> bool:
-    if not old or old.get("epsg") is None or int(old["epsg"]) != epsg:
-        return True
-    keys = ("start_x", "start_y", "end_x", "end_y")
-    return any(old[k] is None or abs(float(old[k]) - new[k]) > SAME_M for k in keys)
+    starts: list[LonLat] = Field(default_factory=list, max_length=MAX_POINTS)
+    ends: list[LonLat] = Field(default_factory=list, max_length=MAX_POINTS)
 
 
 async def _load(project_id: UUID) -> tuple[dict[str, Any], dict[int, dict[str, Any]]]:
@@ -114,44 +65,6 @@ async def _load(project_id: UUID) -> tuple[dict[str, Any], dict[int, dict[str, A
     return project, {int(row["year"]): row for row in rows}
 
 
-async def _store(
-    project_id: UUID, epsg: int, writes: list[dict[str, Any]], drops: list[int]
-) -> None:
-    """바뀐 연도만 넣고(분석 노선 비움) · 범위 밖 연도 행은 지움 — 한 묶음."""
-    async with get_db_pool().acquire() as connection:
-        await connection.begin()
-        try:
-            async with connection.cursor() as cursor:
-                for row in writes:
-                    await cursor.execute(
-                        "INSERT INTO plan_route_years"
-                        " (project_id, year, start_x, start_y, end_x, end_y, epsg)"
-                        " VALUES (%s, %s, %s, %s, %s, %s, %s)"
-                        " ON DUPLICATE KEY UPDATE start_x = VALUES(start_x),"
-                        " start_y = VALUES(start_y), end_x = VALUES(end_x), end_y = VALUES(end_y),"
-                        " epsg = VALUES(epsg), route_json = NULL, fingerprint = NULL,"
-                        " confirmed_at = NULL",
-                        (
-                            str(project_id),
-                            row["year"],
-                            row["start_x"],
-                            row["start_y"],
-                            row["end_x"],
-                            row["end_y"],
-                            epsg,
-                        ),
-                    )
-                for year in drops:
-                    await cursor.execute(
-                        "DELETE FROM plan_route_years WHERE project_id = %s AND year = %s",
-                        (str(project_id), year),
-                    )
-            await connection.commit()
-        except Exception:
-            await connection.rollback()
-            raise
-
-
 def _point(x: Any, y: Any, epsg: Any) -> dict[str, float] | None:
     if x is None or y is None or epsg is None:
         return None
@@ -159,45 +72,51 @@ def _point(x: Any, y: Any, epsg: Any) -> dict[str, float] | None:
     return {"lon": lon, "lat": lat, "x": float(x), "y": float(y)}
 
 
-def base_view(project: dict[str, Any], rows: dict[int, dict[str, Any]]) -> dict[str, Any]:
-    """{epsg, start, end} — 정본 칸 · 비면 첫 연도 시점 + 마지막 연도 종점으로 접음."""
+def _stored(project: dict[str, Any]) -> dict[str, Any] | None:
     stored = project.get("plan_base_json")
-    if isinstance(stored, str):
-        stored = json.loads(stored)
+    return json.loads(stored) if isinstance(stored, str) else stored
+
+
+def base_view(project: dict[str, Any], rows: dict[int, dict[str, Any]]) -> dict[str, Any]:
+    """{epsg, starts, ends} — 정본 칸(옛 한 점 꼴은 감쌈) ·
+    비면 첫 연도 시점 + 마지막 연도 종점으로 접음."""
+    stored = _stored(project)
     if stored:
-        epsg, start, end = stored.get("epsg"), stored.get("start"), stored.get("end")
+        epsg = stored.get("epsg")
+        starts = stored.get("starts", [stored["start"]] if stored.get("start") else [])
+        ends = stored.get("ends", [stored["end"]] if stored.get("end") else [])
     elif rows:
         first, last = rows[min(rows)], rows[max(rows)]
         epsg = first.get("epsg")
-        start = {"x": first.get("start_x"), "y": first.get("start_y")}
-        end = {"x": last.get("end_x"), "y": last.get("end_y")}
+        starts = [{"x": first.get("start_x"), "y": first.get("start_y")}]
+        ends = [{"x": last.get("end_x"), "y": last.get("end_y")}]
     else:
-        epsg, start, end = None, None, None
-    return {
-        "epsg": epsg,
-        "start": _point(start["x"], start["y"], epsg) if start else None,
-        "end": _point(end["x"], end["y"], epsg) if end else None,
-    }
+        epsg, starts, ends = None, [], []
+
+    def points(items: list[dict[str, Any]]) -> list[dict[str, float]]:
+        return [p for p in (_point(i.get("x"), i.get("y"), epsg) for i in items) if p]
+
+    return {"epsg": epsg, "starts": points(starts), "ends": points(ends)}
 
 
 async def load_base(project_id: UUID) -> dict[str, Any]:
-    """전체 시점 · 종점 — D03 이 읽는 길."""
+    """전체 시점 · 종점 목록 — D03 이 읽는 길."""
     project, rows = await _load(project_id)
     return base_view(project, rows)
 
 
-def stored_base(payload: BaseRequest) -> dict[str, Any]:
-    """PUT 몸 → 정본 칸 꼴(TM m)."""
-    given = [p for p in (payload.start, payload.end) if p]
-    epsg = tm_epsg_for_lon(sum(p.lon for p in given) / len(given)) if given else None
+def stored_base(payload: BaseRequest, epsg: int | None = None) -> dict[str, Any]:
+    """PUT 몸 → 정본 칸 꼴(TM m) — 좌표계 = `epsg`(저장된 값) · 없으면 점 평균 경도의 TM 띠."""
+    given = [*payload.starts, *payload.ends]
+    if not given:
+        epsg = None
+    elif epsg is None:
+        epsg = tm_epsg_for_lon(sum(p.lon for p in given) / len(given))
 
-    def tm(p: LonLat | None) -> dict[str, float] | None:
-        if p is None or epsg is None:
-            return None
-        x, y = lonlat_to_tm(p.lon, p.lat, epsg)
-        return {"x": x, "y": y}
+    def tm(items: list[LonLat]) -> list[dict[str, float]]:
+        return [dict(zip(("x", "y"), lonlat_to_tm(p.lon, p.lat, epsg))) for p in items]
 
-    return {"epsg": epsg, "start": tm(payload.start), "end": tm(payload.end)}
+    return {"epsg": epsg, "starts": tm(payload.starts), "ends": tm(payload.ends)}
 
 
 async def _store_base(project_id: UUID, base: dict[str, Any]) -> None:
@@ -221,6 +140,7 @@ async def put_plan_points(
 ) -> dict[str, Any]:
     if not is_forest_service(session):
         raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.")
-    await _load(project_id)  # 없음 · 예정노선 아님 거절
-    await _store_base(project_id, stored_base(payload))
+    project, _ = await _load(project_id)  # 없음 · 예정노선 아님 거절
+    epsg = (_stored(project) or {}).get("epsg")
+    await _store_base(project_id, stored_base(payload, int(epsg) if epsg else None))
     return {"status": "success", **await load_base(project_id)}