feat(M02): 옛 프로젝트 양식 넣기 도구 — 없는 것만 더함 · 폴더 없으면 건너뜀 · 목록 보고 (PLAN 10-5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015LuapLYqN1GGFD8Y1PStD5

M02_MasterTemplete/M02_Template_Migrate.py

@@ -0,0 +1,119 @@
+"""이미 만든 프로젝트에 시스템 양식 넣기 — 한 번 돌리는 도구 (PLAN 10-5).
+
+프로젝트마다 `seed_project(only_missing=True)` — 작업본 · `_initial/` 에 **없는 것만 더함**.
+이미 있는 양식 · 다른 파일은 안 건드림 · 폴더가 없는 프로젝트는 건너뜀(만들지 않음).
+다시 돌려도 됨 — 시스템 양식이 늘었으면 그 몫만 더해짐.
+
+    ./venv/Scripts/python.exe M02_MasterTemplete/M02_Template_Migrate.py [--dry-run] [--report 파일]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from datetime import datetime
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from config import config_system  # noqa: E402
+from config.config_db import close_db_pool, get_db_pool, init_db_pool  # noqa: E402
+from M02_MasterTemplete import M02_Template_Layers as layers  # noqa: E402
+
+
+async def _projects() -> list[dict[str, Any]]:
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        await cursor.execute(
+            """SELECT p.id, p.name, p.company_id, p.user_id, p.storage_path, u.name
+               FROM projects p LEFT JOIN users u ON u.id = p.user_id
+               WHERE p.deleted_at IS NULL ORDER BY p.company_id, p.created_at"""
+        )
+        rows = await cursor.fetchall()
+    keys = ("id", "name", "company_id", "user_id", "storage_path", "owner")
+    return [dict(zip(keys, row, strict=True)) for row in rows]
+
+
+def project_root(storage_path: str | None) -> Path | None:
+    """DB 저장 경로 → 실경로 · 없거나 수상하면 None(폴더를 만들지 않음)."""
+    if not storage_path:
+        return None
+    parts = PurePosixPath(storage_path.replace("\\", "/")).parts
+    if not parts or parts[0] != "storage" or ".." in parts or len(parts) < 2:
+        return None
+    root = Path(os.path.realpath(config_system.STORAGE_BASE_DIR))
+    path = Path(os.path.realpath(root.joinpath(*parts[1:])))
+    if root not in path.parents or not path.is_dir():
+        return None
+    return path
+
+
+def migrate(projects: list[dict[str, Any]], *, dry_run: bool = False) -> list[dict[str, Any]]:
+    """프로젝트마다 넣은 것 `[{…프로젝트, 작업본, 초기, 상태}]`."""
+    system = {f"{row['종류']}/{row['이름']}" for row in layers.list_templates(layers.system_dir())}
+    report = []
+    for project in projects:
+        root = project_root(project.get("storage_path"))
+        entry = {**project, "작업본": [], "초기": [], "상태": ""}
+        if root is None:
+            entry["상태"] = "폴더 없음 — 건너뜀"
+        elif dry_run:
+            have = {
+                f"{row['종류']}/{row['이름']}"
+                for row in layers.list_templates(layers.project_dir(root))
+            }
+            entry["작업본"] = sorted(system - have)
+            entry["상태"] = "넣을 것(시험)"
+        else:
+            done = layers.seed_project(root, only_missing=True)
+            entry.update(작업본=done["작업본"], 초기=done["초기"])
+            entry["상태"] = "넣음" if done["작업본"] or done["초기"] else "이미 있음"
+        report.append(entry)
+    return report
+
+
+def render(report: list[dict[str, Any]], system: list[dict[str, Any]]) -> str:
+    names = " · ".join(f"{row['종류']}/{row['이름']}" for row in system) or "없음"
+    lines = [
+        "# 기존 프로젝트 양식 넣기",
+        "",
+        f"- 돌린 때: {datetime.now().isoformat(timespec='seconds')}",
+        f"- 시스템 양식: {names}",
+        "- 방법: 작업본 · `_initial/` 에 없는 것만 더함 · 있는 양식 · 다른 파일은 그대로",
+        f"- 프로젝트 {len(report)}개 · 넣음 {sum(r['상태'] == '넣음' for r in report)}개",
+        "",
+        "| 회사 | 만든 사람 | 프로젝트 | ID | 넣은 양식 | 상태 |",
+        "| --- | --- | --- | --- | --- | --- |",
+    ]
+    for row in report:
+        added = " · ".join(row["작업본"]) or "—"
+        lines.append(
+            f"| {row['company_id']} | {row.get('owner') or row['user_id']} | {row['name']} "
+            f"| `{row['id']}` | {added} | {row['상태']} |"
+        )
+    return "\n".join(lines) + "\n"
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--report", type=Path)
+    args = parser.parse_args()
+    await init_db_pool()
+    try:
+        projects = await _projects()
+    finally:
+        await close_db_pool()
+    report = migrate(projects, dry_run=args.dry_run)
+    text = render(report, layers.list_templates(layers.system_dir()))
+    if args.report:
+        args.report.write_text(text, encoding="utf-8")
+    sys.stdout.reconfigure(encoding="utf-8")
+    print(text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

resources/tester/test_m02_template_layers.py

@@ -220,3 +220,48 @@ def test_프로젝트_만들기_자리에서_시스템_양식을_복사(world: d
     assert got["문서"] == {"판": 1, "열": []}  # 회사 공식이 있어도 시스템
     assert (root / "templates/_initial/drawing/A1_도각.json").is_file()
     assert (root / "project_manifest.json").is_file()
+
+
+def test_옛_프로젝트_넣기_도구는_더하기만_폴더_없으면_건너뜀(world: dict[str, Any]) -> None:
+    from M02_MasterTemplete import M02_Template_Migrate as migrate
+
+    old = world["storage"] / "7/42/old-project"
+    (old / "B05_Profile").mkdir(parents=True)
+    (old / "B05_Profile/keep.txt").write_text("x", encoding="utf-8")
+    root1 = _root(world, P1)
+    layers.write_template(root1 / "templates", "table", "구조물집계표", {"고침": 1})
+    (root1 / "templates/drawing/A1_도각.json").unlink()
+    projects = [
+        {
+            "id": "old",
+            "name": "옛",
+            "company_id": 7,
+            "user_id": 42,
+            "storage_path": "storage/7/42/old-project",
+        },
+        {
+            "id": P1,
+            "name": "첫째",
+            "company_id": 7,
+            "user_id": 42,
+            "storage_path": f"storage/7/42/{P1}",
+        },
+        {
+            "id": "gone",
+            "name": "없음",
+            "company_id": 7,
+            "user_id": 42,
+            "storage_path": "storage/7/42/gone",
+        },
+    ]
+    dry = migrate.migrate(projects, dry_run=True)
+    assert not (old / "templates").exists() and dry[0]["상태"] == "넣을 것(시험)"
+    report = {row["id"]: row for row in migrate.migrate(projects)}
+    assert report["old"]["상태"] == "넣음" and (old / "templates/_initial/table").is_dir()
+    assert (old / "B05_Profile/keep.txt").read_text(encoding="utf-8") == "x"
+    assert report[P1]["작업본"] == ["drawing/A1_도각"]
+    assert layers.read_template(root1 / "templates", "table", "구조물집계표")["문서"] == {"고침": 1}
+    assert report["gone"]["상태"].startswith("폴더 없음")
+    assert not (world["storage"] / "7/42/gone").exists()
+    assert migrate.migrate(projects)[0]["상태"] == "이미 있음"
+    assert "| 7 | 42 | 옛 |" in migrate.render(list(report.values()), [])