feat(765-6): 초기값 사본에서 배수 공유 캐시 뺌 · 복원은 노선 몫만 갈음(캐시 제자리) · 복제가 설계 벌 묶음 안 지표면 id 고침

- common_util_initial_snapshot: 배수 사본 = 노선 몫만(하강장 · 도엽 좌표 · 강우 뺌 · 약 105 → 20 MB) · 복원은 폴더 통째 지우지 않음 · 캐시 든 옛 사본의 캐시 칸은 건너뜀
- common_util_project_copy._fix_files(통째 복제 · --replace 같은 길): design_sets/<지형>/ 의 db.json · 그 벌 초기값 · B06 표본 설정 · meta.json 지표면 id → 새 id · 노선 id 는 꺼낼 때 늘 새로 받아 안 고침 · 코리도 안 건드림
- 시험 test_765_design_sets 에 둘 덧

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

common_util/common_util_initial_snapshot.py

@@ -169,6 +169,49 @@ def _copy_tree(source: Path, target: Path) -> None:
     shutil.copytree(source, target)
 
 
+_DRAINAGE_TREE = "B04_PreProcess/drainage"
+
+
+def _drainage_shared(name: str) -> bool:
+    return name.startswith(DRAINAGE_SHARED_PREFIXES)
+
+
+def _save_drainage(source: Path, target: Path) -> None:
+    """배수 폴더 중 노선 몫만 뜸 — 공유 캐시(하강장 · 도엽 좌표 · 강우)는 뺌(765-6 ·
+    약 105 → 20 MB)."""
+    if not source.is_dir():
+        return
+    if target.exists():
+        shutil.rmtree(target)
+    top = str(source)
+    shutil.copytree(
+        source,
+        target,
+        ignore=lambda folder, names: [n for n in names if folder == top and _drainage_shared(n)],
+    )
+
+
+def _restore_drainage(stored: Path, live: Path) -> None:
+    """노선 몫만 갈음 — 공유 캐시는 제자리에 둠(지우면 다음 배수 계산이 처음부터 · 765-6) ·
+    캐시 든 옛 사본도 캐시 칸은 건너뜀."""
+    if live.is_dir():
+        for path in live.iterdir():
+            if _drainage_shared(path.name):
+                continue
+            if path.is_dir():
+                shutil.rmtree(path)
+            else:
+                path.unlink()
+    live.mkdir(parents=True, exist_ok=True)
+    for path in stored.iterdir():
+        if _drainage_shared(path.name):
+            continue
+        if path.is_dir():
+            shutil.copytree(path, live / path.name)
+        else:
+            shutil.copy2(path, live / path.name)
+
+
 async def _dump_rows(
     connection: aiomysql.Connection, table: str, route_id: int
 ) -> list[dict[str, Any]]:
@@ -211,7 +254,8 @@ async def save_initial_snapshot(
         _write_design_route_csv(target / DESIGN_ROUTE_CSV_NAME, design_route_points)
 
     for tree in _FILE_TREES:
-        _copy_tree(root / tree, target / tree.replace("/", "__"))
+        save = _save_drainage if tree == _DRAINAGE_TREE else _copy_tree
+        save(root / tree, target / tree.replace("/", "__"))
 
     # 코리도는 체인이 마지막에 만든다 — 있으면 함께 뜬다(없으면 브라우저 폴백 그대로).
     from B05_Profile.B05_Profile_Router_Corridor import corridor_path
@@ -294,6 +338,9 @@ def restore_snapshot_files(project_root: Path, route_id: int | None = None) -> N
             else:  # 배수 폴더 없는 스냅샷 — 관 편집은 비움(초기값에 없던 것)
                 shutil.rmtree(edits, ignore_errors=True)
             continue
+        if tree == _DRAINAGE_TREE:
+            _restore_drainage(stored, root / tree)
+            continue
         _copy_tree(stored, root / tree)
 
     # 코리도는 복원으로 만든 **새 route id** 이름으로 되돌린다 — 이름이 어긋나면 브라우저가

common_util/common_util_project_copy.py

@@ -158,18 +158,68 @@ def _fix_files(
     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():
-        dump = json.loads(snapshot.read_text(encoding="utf-8"))
-        changed = False
-        for route in dump.get("routes", []):
-            old_model = route.get("surface_model_id")
-            if old_model in model_map:
-                route["surface_model_id"] = model_map[old_model]
-                changed = True
-        if changed:
-            snapshot.write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8")
-        print("초기값 스냅샷 지표면 id 교정:", changed)
+    print(
+        "초기값 스냅샷 지표면 id 교정:", _fix_dump_models(new_root / "initial_snapshot", model_map)
+    )
+    print("설계 벌 묶음 지표면 id 교정:", _fix_design_sets(new_root, model_map))
+
+
+def _fix_dump_models(folder: Path, model_map: dict) -> bool:
+    """`folder/db.json` 노선 행의 지표면 id → 새 id · 바꿨으면 참."""
+    path = folder / "db.json"
+    if not path.is_file():
+        return False
+    dump = json.loads(path.read_text(encoding="utf-8"))
+    changed = False
+    for route in dump.get("routes", []):
+        old_model = route.get("surface_model_id")
+        if old_model in model_map:
+            route["surface_model_id"] = model_map[old_model]
+            changed = True
+    if changed:
+        path.write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8")
+    return changed
+
+
+def _fix_sampling(root: Path, model_map: dict) -> bool:
+    """`B06_Section/sampling.json` 의 지표면 id → 새 id(코리도는 안 건드림)."""
+    path = root / "B06_Section" / "sampling.json"
+    if not path.is_file():
+        return False
+    data = json.loads(path.read_text(encoding="utf-8"))
+    if data.get("surface_model_id") not in model_map:
+        return False
+    data["surface_model_id"] = model_map[data["surface_model_id"]]
+    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+    return True
+
+
+def _fix_design_sets(root: Path, model_map: dict) -> dict[str, list[str]]:
+    """쉬는 설계 벌 묶음(765-6) 안 지표면 id → 새 id — db.json · 그 벌 초기값 · B06 표본 설정 ·
+    meta.json. 노선 id 는 꺼낼 때 늘 새로 받아(묶음 안에서만 짝) 안 고침 → 묶음 → 고친 자리."""
+    from common_util.common_util_design_sets import sets_dir
+
+    report: dict[str, list[str]] = {}
+    base = sets_dir(root)
+    for bundle in sorted(base.iterdir()) if base.is_dir() else []:
+        fixed = [
+            name
+            for name, done in (
+                ("db.json", _fix_dump_models(bundle, model_map)),
+                ("initial_snapshot", _fix_dump_models(bundle / "initial_snapshot", model_map)),
+                ("sampling", _fix_sampling(bundle / "files", model_map)),
+            )
+            if done
+        ]
+        meta_path = bundle / "meta.json"
+        if meta_path.is_file():
+            meta = json.loads(meta_path.read_text(encoding="utf-8"))
+            if meta.get("model_id") in model_map:
+                meta["model_id"] = model_map[meta["model_id"]]
+                meta_path.write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8")
+                fixed.append("meta.json")
+        report[bundle.name] = fixed
+    return report
 
 
 def _write_manifest(root: Path, manifest: dict[str, Any]) -> None:

resources/tester/test_765_design_sets.py

@@ -517,3 +517,53 @@ def test_비우기에_묶음_포함(tmp_path: Path) -> None:
     removed = purge_output_folders(tmp_path, keep_preprocess=True)
     assert SETS_DIR in removed and not (tmp_path / SETS_DIR).exists()
     assert (tmp_path / "B04_PreProcess" / "models").is_dir()
+
+
+# ── 765-6 초기값 사본 = 캐시 뺌 · 복제가 묶음 안 지표면 id 고침 ────────────
+
+
+def test_초기값_사본은_배수_캐시를_빼고_복원은_캐시를_안_지움(tmp_path: Path) -> None:
+    from common_util import common_util_initial_snapshot as snap
+
+    _seed_files(tmp_path)
+    drainage = tmp_path / "B04_PreProcess" / "drainage"
+    (drainage / "rainfall_stations").mkdir()
+    (drainage / "rainfall_stations" / "a.json").write_text("st", encoding="utf-8")
+    target = tmp_path / "initial_snapshot" / "B04_PreProcess__drainage"
+    snap._save_drainage(drainage, target)
+    saved = sorted(p.name for p in target.iterdir())
+    assert saved == ["01_primary_region.geojson", "edits", "manifest.json"]  # 캐시 없음
+
+    # 작업본이 바뀜 · 캐시는 새로 지어짐 · 옛 사본(캐시 듦) 꼴도 섞음
+    (drainage / "01_primary_region.geojson").write_text("edited", encoding="utf-8")
+    (drainage / "04_detailed_basins.geojson").write_text("new", encoding="utf-8")
+    (drainage / "descent_ab_cd.npz").write_text("d2", encoding="utf-8")
+    (target / "00_sheet_metric.pkl").write_text("old-cache", encoding="utf-8")
+    snap._restore_drainage(target, drainage)
+    live = {p.name: p for p in drainage.iterdir()}
+    assert live["01_primary_region.geojson"].read_text(encoding="utf-8") == "{}"
+    assert "04_detailed_basins.geojson" not in live  # 초기값에 없던 노선 몫 = 지움
+    assert live["descent_ab_cd.npz"].read_text(encoding="utf-8") == "d2"  # 캐시 그대로
+    assert live["00_sheet_metric.pkl"].read_text(encoding="utf-8") == "s"  # 옛 사본 캐시 안 덮음
+    assert (live["rainfall_stations"] / "a.json").is_file()
+    assert (live["edits"] / "pipe_points.json").is_file()
+
+
+def test_복제가_묶음_안_지표면_id_고침(tmp_path: Path) -> None:
+    from common_util.common_util_project_copy import _fix_design_sets
+
+    db = _seed_db()
+    _seed_files(tmp_path)
+    asyncio.run(sets.park_active_set(_Conn(db), tmp_path, PID, FP))
+    bundle = tmp_path / SETS_DIR / "las"
+    corridor = bundle / "files" / "B05_Profile" / "corridor" / "corridor_0011.json"
+    report = _fix_design_sets(tmp_path, {5: 905})
+    assert sorted(report["las"]) == ["db.json", "meta.json", "sampling"]
+    dump = json.loads((bundle / "db.json").read_text(encoding="utf-8"))
+    assert {r["surface_model_id"] for r in dump["routes"]} == {905}
+    assert [r["id"] for r in dump["routes"]] == [10, 11]  # 노선 id 는 꺼낼 때 새로
+    assert sets.read_meta(tmp_path, "las")["model_id"] == 905
+    sampling = bundle / "files" / "B06_Section" / "sampling.json"
+    assert json.loads(sampling.read_text(encoding="utf-8")) == {"surface_model_id": 905}
+    assert corridor.is_file()  # 코리도는 안 건드림
+    assert sets.bundle_ready(tmp_path, "las", FP)  # 지문은 id 를 안 봄