fix(761-5): 복제 · 바꿔 넣기 뒤 노선 · 지표면 id 박힌 storage 파일 새 id 로

- common_util_project_copy.rekey_id_files: 코리도 이름 corridor_{노선 id:04d} → 새 노선 id(겹침 대비 임시 이름 거침) · 짝 없는 옛 코리도 지움 · B06_Section/sampling.json surface_model_id → 새 지표면 id
- _insert_rows · _insert_full 이 (지표면 · 노선) id 짝을 함께 돌려줌 · full 복제 · --replace · --storage-only 셋 다 _fix_files 에서 rekey
- id_maps_by_order: 이미 복제된 두 프로젝트의 id 짝(차례 · 행 수 다르면 ValueError) — storage-only 가 씀
- storage 경로 함수 전수: id 박힌 이름 = 코리도뿐(B08 이정표 그림 이름 = 거래처 상자 id · 추천 저장본 = 해시) · 내용 = sampling.json 지표면 id(초기값 스냅샷 지표면 id 는 종전대로)
- 지킴 시험 test_761_2: 노선 id 짝 단정 · rekey(이름 겹침 · 짝 없음 지움 · sampling)
- 바꿔 넣은 B 13 개 고침(tmp 스크립트 · 차례 짝 + 내용 칸 대조): 코리도 셋씩 새 id · corridor_0370 지움 · sampling 911 → 각 새 id

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

common_util/common_util_project_copy.py

@@ -116,8 +116,47 @@ def replace_project_id(root: Path, source_id: str, new_id: str) -> list[str]:
     return replaced
 
 
-def _fix_files(new_root: Path, source_id: str, new_id: str, model_map: dict) -> None:
+def rekey_id_files(root: Path, model_map: dict, route_map: dict) -> dict[str, Any]:
+    """다시 받은 id 가 박힌 파일을 새 id 로 — 코리도 이름 `corridor_{노선 id:04d}` (짝 없는 옛
+    이름 = 지움) · `B06_Section/sampling.json` 의 surface_model_id → 바꾼 것 보고(761-5)."""
+    from B05_Profile.B05_Profile_Router_Corridor import corridor_path
+
+    report: dict[str, Any] = {"corridor": {}, "corridor_removed": [], "sampling": None}
+    directory = root / "B05_Profile" / "corridor"
+    moving = []
+    for path in sorted(directory.glob("corridor_*.json")) if directory.is_dir() else []:
+        try:
+            old = int(path.stem.removeprefix("corridor_"))
+        except ValueError:
+            continue
+        if old in route_map:
+            # 이름이 옛 · 새 사이에 겹칠 수 있어 임시 이름을 거침
+            temp = path.with_name(f".rekey_{path.name}")
+            path.replace(temp)
+            moving.append((temp, old))
+        else:
+            path.unlink()
+            report["corridor_removed"].append(path.name)
+    for temp, old in moving:
+        temp.replace(corridor_path(root, route_map[old]))
+        report["corridor"][old] = route_map[old]
+
+    sampling = root / "B06_Section" / "sampling.json"
+    if sampling.is_file():
+        data = json.loads(sampling.read_text(encoding="utf-8"))
+        old_model = data.get("surface_model_id")
+        if old_model in model_map:
+            data["surface_model_id"] = model_map[old_model]
+            sampling.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+            report["sampling"] = [old_model, model_map[old_model]]
+    return report
+
+
+def _fix_files(
+    new_root: Path, source_id: str, new_id: str, model_map: dict, route_map: dict
+) -> None:
     print("프로젝트 id 치환:", replace_project_id(new_root, source_id, new_id))
+    print("노선 · 지표면 id 파일:", rekey_id_files(new_root, model_map, route_map))
 
     snapshot = new_root / "initial_snapshot" / "db.json"
     if snapshot.is_file():
@@ -245,8 +284,8 @@ async def _route_children(cursor, route_ids: list[int]) -> dict[str, list]:
 
 async def _insert_full(
     connection, rows: dict[str, list], new_id: str, new_name: str, new_storage: str, owner, members
-) -> dict[str, int]:
-    """DB 행을 새 id 로 다시 이어 한 트랜잭션에 넣음 → 지표면 id 짝(옛 → 새)."""
+) -> tuple[dict[int, int], dict[int, int]]:
+    """DB 행을 새 id 로 다시 이어 한 트랜잭션에 넣음 → (지표면 · 노선) id 짝(옛 → 새)."""
     await connection.begin()
     async with connection.cursor(aiomysql.DictCursor) as cursor:
         project = dict(rows["projects"][0])
@@ -264,7 +303,7 @@ async def _insert_full(
         )
         for user in member_ids:
             await _insert(cursor, "project_members", {"project_id": new_id, "user_id": user})
-        model_map = await _insert_rows(cursor, rows, new_id)
+        model_map, route_map = await _insert_rows(cursor, rows, new_id)
     await connection.commit()
     print(
         f"DB 복제 완료: 참여자 {len(member_ids)} · 입력파일 {len(rows['input_files'])}"
@@ -274,12 +313,14 @@ async def _insert_full(
         f" · 단계 {len(rows['project_workflow_stages'])}"
         f" · 예정노선 {' · '.join(f'{t} {len(rows[t])}' for t in _PLAN_ALL)}"
     )
-    return model_map
+    return model_map, route_map
 
 
-async def _insert_rows(cursor, rows: dict[str, list], new_id: str) -> dict[int, int]:
+async def _insert_rows(
+    cursor, rows: dict[str, list], new_id: str
+) -> tuple[dict[int, int], dict[int, int]]:
     """projects · project_members 밖 복제 표 행을 new_id 로 넣고 자동증가 id 를 다시 이음 →
-    지표면 id 짝(옛 → 새) · 트랜잭션은 부르는 쪽 몫(full 복제 · 자리 바꿔 넣기)."""
+    (지표면 · 노선) id 짝(옛 → 새) · 트랜잭션은 부르는 쪽 몫(full 복제 · 자리 바꿔 넣기)."""
     file_map: dict[int, int] = {}
     for row in rows["input_files"]:
         row = dict(row)
@@ -361,7 +402,7 @@ async def _insert_rows(cursor, rows: dict[str, list], new_id: str) -> dict[int,
         # 업체가 이은 임도는 원본 D 몫 — 복제본은 잇지 않은 채로 시작
         row["linked_project_id"] = None
         await _insert(cursor, "plan_route_assignees", row)
-    return model_map
+    return model_map, route_map
 
 
 async def clone_full(
@@ -389,9 +430,11 @@ async def clone_full(
     # 1) 파일 먼저 - 실패하면 DB 를 건드리지 않은 채로 끝난다.
     _copy_folder(source_root, new_root)
     # 2) DB - 한 트랜잭션으로.
-    model_map = await _insert_full(connection, rows, new_id, new_name, new_storage, owner, members)
-    # 3) 파일 안에 박힌 옛 프로젝트 id 치환 + 초기값 스냅샷의 지표면 모델 id 교정.
-    _fix_files(new_root, source_id, new_id, model_map)
+    model_map, route_map = await _insert_full(
+        connection, rows, new_id, new_name, new_storage, owner, members
+    )
+    # 3) 파일 안에 박힌 옛 프로젝트 id 치환 + 초기값 스냅샷의 지표면 모델 id 교정 + 노선 id 파일.
+    _fix_files(new_root, source_id, new_id, model_map, route_map)
     _write_manifest(
         new_root,
         {
@@ -409,29 +452,41 @@ async def clone_full(
     return new_id
 
 
+async def id_maps_by_order(
+    connection, source_id: str, new_id: str
+) -> tuple[dict[int, int], dict[int, int]]:
+    """이미 복제된 두 프로젝트의 (지표면 · 노선) id 짝 — 복제는 원본 id 차례대로 넣으므로 차례로
+    짝 지음 · 행 수가 다르면 ValueError(복제 뒤 한쪽이 바뀜)."""
+    maps = []
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        for table in ("surface_models", "routes"):
+            ids = {}
+            for project_id in (source_id, new_id):
+                rows = await _rows(
+                    cursor,
+                    f"SELECT id FROM {table} WHERE project_id = %s ORDER BY id",
+                    (project_id,),
+                )
+                ids[project_id] = [item["id"] for item in rows]
+            maps.append(dict(zip(ids[source_id], ids[new_id], strict=True)))
+    return maps[0], maps[1]
+
+
 async def clone_storage_only(connection, source_id: str, new_id: str) -> None:
     """DB 는 SELECT 만 — 원본 · 새 프로젝트 행(이미 복제됨)으로 폴더 자리 · 지표면 id 짝."""
     paths = {}
-    models = {}
     async with connection.cursor(aiomysql.DictCursor) as cursor:
         for project_id in (source_id, new_id):
             row = await _rows(
                 cursor, "SELECT storage_path FROM projects WHERE id = %s", (project_id,)
             )
             paths[project_id] = row[0]["storage_path"]
-            rows = await _rows(
-                cursor,
-                "SELECT id FROM surface_models WHERE project_id = %s ORDER BY id",
-                (project_id,),
-            )
-            models[project_id] = [item["id"] for item in rows]
-    # 복제는 지표면 행을 원본 id 차례대로 넣는다 — 차례로 짝 지음.
-    model_map = dict(zip(models[source_id], models[new_id], strict=True))
+    model_map, route_map = await id_maps_by_order(connection, source_id, new_id)
     new_root = Path(resolve_stored_project_path(paths[new_id]))
     if any(path.is_file() for path in new_root.rglob("*")):
         raise RuntimeError(f"새 폴더에 파일이 이미 있음: {new_root}")
     _copy_folder(Path(resolve_stored_project_path(paths[source_id])), new_root)
-    _fix_files(new_root, source_id, new_id, model_map)
+    _fix_files(new_root, source_id, new_id, model_map, route_map)
     print(f"storage 만 복제 완료: {new_root}")
 
 

common_util/common_util_project_copy_replace.py

@@ -137,13 +137,13 @@ async def replace_project(
             await cursor.execute(
                 f"UPDATE projects SET {sets} WHERE id = %s", (*values.values(), target_id)
             )
-            model_map = await _insert_rows(cursor, rows, target_id)
+            model_map, route_map = await _insert_rows(cursor, rows, target_id)
         await connection.commit()
     except Exception:
         await connection.rollback()
         raise
-    # 3) 파일 안 옛 프로젝트 id · 초기값 스냅샷 지표면 id
-    _fix_files(target_root, source_id, target_id, model_map)
+    # 3) 파일 안 옛 프로젝트 id · 초기값 스냅샷 지표면 id · 노선 id 이름 파일
+    _fix_files(target_root, source_id, target_id, model_map, route_map)
     tables = {name: len(items) for name, items in rows.items()}
     _write_manifest(
         target_root,

resources/tester/test_761_2_clone_replace.py

@@ -3,6 +3,7 @@
 - 유지 칸: 대상 id · 이름 · 주인 · 회사 · 종류 · 저장 자리 · 담당자는 대상 값 · 나머지는 원본
 - 지우는 표 목록 = 복제 표 목록(projects · project_members 뺌)과 같음 — 표가 늘면 여기서 빨강
 - id 다시 잇기: 새 행이 새 노선 · 새 지표면 · 새 입력 파일 id 를 가리킴 · 단계 설정 id 도
+- 노선 · 지표면 id 가 박힌 storage 파일(코리도 이름 · sampling.json) → 새 id · 짝 없는 코리도 지움(761-5)
 """
 
 import asyncio
@@ -98,7 +99,7 @@ def test_insert_rows_relinks_ids():
         **{table: [] for table in cp._PLAN_ALL},
     }
     cursor = _Cursor()
-    model_map = asyncio.run(cp._insert_rows(cursor, rows, "dst"))
+    model_map, route_map = asyncio.run(cp._insert_rows(cursor, rows, "dst"))
     new = {table: row for table, row in cursor.inserts}
     assert [table for table, _ in cursor.inserts][0] == "input_files"
     assert "projects" not in new and "project_members" not in new
@@ -112,7 +113,37 @@ def test_insert_rows_relinks_ids():
     assert new["terrain_layers"]["surface_model_id"] == model_id
     assert new["routes"]["surface_model_id"] == model_id
     route_id = 1005
+    assert route_map == {5: route_id}
     assert new["route_points"]["route_id"] == route_id
     assert new["longitudinal_sections"]["route_id"] == route_id
     params = json.loads(new["project_workflow_stages"]["params"])
     assert params == {"input_file_id": str(file_id), "surface_model_id": model_id}
+
+
+def test_rekey_id_files_moves_route_and_model_ids(tmp_path):
+    corridor = tmp_path / "B05_Profile" / "corridor"
+    corridor.mkdir(parents=True)
+    # 옛 1068 → 새 1069 · 옛 1069 → 새 1091 (이름 겹침) · 0370 = 짝 없음
+    for old in (370, 1068, 1069):
+        (corridor / f"corridor_{old:04d}.json").write_text(str(old), encoding="utf-8")
+    (corridor / "corridor_x.json").write_text("x", encoding="utf-8")
+    sampling = tmp_path / "B06_Section" / "sampling.json"
+    sampling.parent.mkdir(parents=True)
+    sampling.write_text(json.dumps({"filter_key": "classification", "surface_model_id": 911}))
+
+    report = cp.rekey_id_files(tmp_path, {911: 952}, {1068: 1069, 1069: 1091})
+
+    names = {path.name: path.read_text(encoding="utf-8") for path in corridor.iterdir()}
+    assert names == {
+        "corridor_1069.json": "1068",
+        "corridor_1091.json": "1069",
+        "corridor_x.json": "x",
+    }
+    assert report["corridor_removed"] == ["corridor_0370.json"]
+    assert json.loads(sampling.read_text())["surface_model_id"] == 952
+    # 폴더 없음 = 빈 보고
+    assert cp.rekey_id_files(tmp_path / "없음", {}, {}) == {
+        "corridor": {},
+        "corridor_removed": [],
+        "sampling": None,
+    }