diff --git a/M02_MasterTemplete/M02_Template_Migrate.py b/M02_MasterTemplete/M02_Template_Migrate.py new file mode 100644 index 00000000..652fe2ae --- /dev/null +++ b/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()) diff --git a/resources/tester/test_m02_template_layers.py b/resources/tester/test_m02_template_layers.py index c8a5f125..5c71d0ca 100644 --- a/resources/tester/test_m02_template_layers.py +++ b/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()), [])