Aislo/M02_MasterTemplete/M02_Template_Migrate.py

"""이미 만든 프로젝트에 시스템 양식 넣기 — 한 번 돌리는 도구 (PLAN 10-5).

프로젝트마다 `seed_project(only_missing=True)` — 작업본 · `_initial/` 에 **없는 것만 더함**.
이미 있는 양식 · 다른 파일은 안 건드림 · 폴더가 없는 프로젝트는 건너뜀(만들지 않음).
다시 돌려도 됨 — 시스템 양식이 늘었으면 그 몫만 더해짐.
`--refresh-tables` — 옛 표 틀(계산 열에 바인딩)만 새 판으로 · 손 값은 지킴.

    ./venv/Scripts/python.exe M02_MasterTemplete/M02_Template_Migrate.py \\
        [--dry-run] [--refresh-tables] [--report 파일]
"""

from __future__ import annotations

import argparse
import asyncio
import copy
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_Table_Fill as fill  # 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 _stale_frame(document: Any) -> bool:
    """옛 표 틀 — 계산 열에 바인딩(설계값 딱지)이 붙은 판."""
    return isinstance(document, dict) and any(
        isinstance(column, dict)
        and column.get("바인딩")
        and (column.get("식") or column["바인딩"].get("값") == "식")
        for column in document.get("열") or []
    )


def refresh_tables(root: Path, *, dry_run: bool = False) -> list[str]:
    """옛 표 틀을 시스템 새 판으로 — 작업본 · `_initial/` 둘 다 · 줄(손 값) · 변수 · 보기는 지킴.

    손댄 것이 없으면 시스템 파일을 그대로 복사(판이 시스템과 같음).
    """
    done: list[str] = []
    for row in layers.list_templates(layers.system_dir()):
        if row["종류"] != "table":
            continue
        system = layers.read_template(layers.system_dir(), "table", row["이름"])
        for folder in (layers.project_dir(root), layers.initial_dir(root)):
            got = layers.read_template(folder, "table", row["이름"])
            if not system or not got or not _stale_frame(got["문서"]):
                continue
            done.append(f"{folder.name}/table/{row['이름']}")
            if dry_run:
                continue
            old = fill.strip_design(got["문서"])
            doc = copy.deepcopy(system["문서"])
            doc["변수"] = {**(doc.get("변수") or {}), **(old.get("변수") or {})}
            doc.update({key: old[key] for key in ("줄", "보기") if key in old})
            if doc == system["문서"]:
                layers.copy_templates(
                    layers.system_dir(),
                    folder,
                    source_layer="system",
                    kind="table",
                    name=row["이름"],
                )
            else:
                layers.write_template(folder, "table", row["이름"], doc)
                layers.note_source(folder, "table", row["이름"], "system", system["판"])
    return done


def migrate(
    projects: list[dict[str, Any]], *, dry_run: bool = False, refresh: 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 not None and refresh:
            entry["표갱신"] = refresh_tables(root, dry_run=dry_run)
        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} | {' · '.join(row['표갱신']) or '—'} | {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("--refresh-tables", 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, refresh=args.refresh_tables)
    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())