Aislo/M02_MasterTemplete/M02_Template_Migrate.py

"""이미 만든 프로젝트에 시스템 양식 넣기 — 한 번 돌리는 도구 (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())