feat(M02): 양식 층 길 — 층 읽기·쓰기 · 초기화 · 적용 · 저장 · 가져올 것 · 채운 표 · 층마다 권한 (PLAN 10-5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015LuapLYqN1GGFD8Y1PStD5
2026-09-25 09:36:59 +09:00
parent a7dbb1f776
commit 7d01d62f1b
2 changed files with 600 additions and 0 deletions
@@ -0,0 +1,417 @@
+"""M02 양식 층 API — 계약 `tmp/M02_분석/6_계약.md` 「층 (sub4)」.
+
+⚠ 등록은 `main.py`(로그인만) · 층마다 권한은 여기서:
+    system    읽기만(고치기는 `/api/m02/templates` 시스템 관리자 길)
+    company   같은 회사 읽기 · 쓰기는 회사 관리자(ADMIN · 마스터 · 시스템 관리자)
+    personal  본인 읽기·쓰기 · 같은 회사 사람 것은 읽기만(가져오기)
+    project   같은 회사 프로젝트 · 작업본 쓰기 · `_initial/` 은 안 씀
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any, Literal
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel, ConfigDict, Field
+
+from common_util.common_util_auth import verify_session
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_db import get_db_pool
+from M02_MasterTemplete import M02_Template_Layers as layers
+
+router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Layers"])
+
+Layer = Literal["system", "company", "personal", "project"]
+
+
+class SaveBody(BaseModel):
+    판: str | None = None
+    문서: dict[str, Any]
+
+
+class TargetBody(BaseModel):
+    종류: str | None = None
+    이름: str | None = None
+
+
+class ApplyBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    from_: Layer = Field(alias="from")
+    user_id: int | None = None
+    project_id: str | None = None
+    종류: str | None = None
+    이름: str | None = None
+
+
+class SaveAsBody(BaseModel):
+    to: Literal["personal", "company"]
+    종류: str
+    이름: str
+
+
+# ── DB (시험이 바꿔 끼움) ─────────────────────────────
+
+
+async def _project_row(project_id: str) -> dict[str, Any] | None:
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        await cursor.execute(
+            """SELECT id, name, company_id, user_id, storage_path FROM projects
+               WHERE id = %s AND deleted_at IS NULL""",
+            (str(project_id),),
+        )
+        row = await cursor.fetchone()
+    if not row:
+        return None
+    keys = ("id", "name", "company_id", "user_id", "storage_path")
+    return dict(zip(keys, row, strict=True))
+
+
+async def _user_company(user_id: int) -> int | None:
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        await cursor.execute(
+            "SELECT company_id FROM users WHERE id = %s AND deleted_at IS NULL", (user_id,)
+        )
+        row = await cursor.fetchone()
+    return row[0] if row else None
+
+
+async def _company_users(company_id: int) -> list[dict[str, Any]]:
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        await cursor.execute(
+            "SELECT id, name FROM users WHERE company_id = %s AND deleted_at IS NULL ORDER BY id",
+            (company_id,),
+        )
+        rows = await cursor.fetchall()
+    return [{"user_id": row[0], "name": row[1]} for row in rows]
+
+
+async def _company_projects(company_id: int) -> list[dict[str, Any]]:
+    pool = get_db_pool()
+    async with pool.acquire() as connection, connection.cursor() as cursor:
+        await cursor.execute(
+            """SELECT id, name, storage_path FROM projects
+               WHERE company_id = %s AND deleted_at IS NULL ORDER BY created_at DESC""",
+            (company_id,),
+        )
+        rows = await cursor.fetchall()
+    return [{"project_id": str(row[0]), "name": row[1], "storage_path": row[2]} for row in rows]
+
+
+# ── 권한 · 자리 ───────────────────────────────────────
+
+
+def _is_system_admin(session: dict[str, Any]) -> bool:
+    return session.get("role") == "SYSTEM_ADMIN"
+
+
+def _is_company_admin(session: dict[str, Any]) -> bool:
+    return session.get("role") in ("ADMIN", "SYSTEM_ADMIN") or bool(session.get("is_master"))
+
+
+async def _project(session: dict[str, Any], project_id: str | None) -> dict[str, Any]:
+    """같은 회사 프로젝트 한 건 + `root`(실경로). 아니면 403/404."""
+    if not project_id:
+        raise HTTPException(status_code=400, detail="project_id 가 필요합니다.")
+    row = await _project_row(project_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
+    if not _is_system_admin(session) and row["company_id"] != session.get("company_id"):
+        raise HTTPException(status_code=403, detail="다른 회사의 프로젝트입니다.")
+    if not row.get("storage_path"):
+        raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
+    row["root"] = Path(resolve_stored_project_path(row["storage_path"]))
+    return row
+
+
+def _company_of(session: dict[str, Any], project: dict[str, Any] | None) -> int:
+    company_id = project["company_id"] if project else session.get("company_id")
+    if company_id is None:
+        raise HTTPException(status_code=403, detail="회사 연결이 필요합니다.")
+    return int(company_id)
+
+
+async def _same_company_user(session: dict[str, Any], user_id: int, company_id: int) -> None:
+    if user_id == session.get("user_id"):
+        return
+    if await _user_company(user_id) != company_id:
+        raise HTTPException(status_code=403, detail="같은 회사 사람의 양식만 볼 수 있습니다.")
+
+
+async def _layer_dir(
+    session: dict[str, Any],
+    layer: str,
+    *,
+    project_id: str | None,
+    user_id: int | None = None,
+    write: bool = False,
+) -> Path:
+    """층 폴더 — 읽기·쓰기 권한까지 여기서 가름."""
+    if layer == "system":
+        if write:
+            raise HTTPException(status_code=403, detail="시스템 양식은 마스터 템플릿 화면에서만.")
+        return layers.system_dir()
+    if layer == "project":
+        return layers.project_dir((await _project(session, project_id))["root"])
+    project = await _project(session, project_id) if project_id else None
+    company_id = _company_of(session, project)
+    if layer == "company":
+        if write and not _is_company_admin(session):
+            raise HTTPException(status_code=403, detail="회사 공식 양식은 회사 관리자만.")
+        return layers.company_dir(company_id)
+    if layer == "personal":
+        owner = int(user_id) if user_id is not None else int(session["user_id"])
+        if write and owner != session.get("user_id"):
+            raise HTTPException(status_code=403, detail="개인 양식은 본인만 고칩니다.")
+        await _same_company_user(session, owner, company_id)
+        return layers.personal_dir(company_id, owner)
+    raise HTTPException(status_code=404, detail="없는 층입니다.")
+
+
+def _bad(error: ValueError) -> HTTPException:
+    return HTTPException(status_code=400, detail=str(error))
+
+
+# ── 층 길 ─────────────────────────────────────────────
+
+
+@router.get("/layers/{layer}/templates")
+async def list_layer(
+    layer: Layer,
+    project_id: str | None = Query(None),
+    user_id: int | None = Query(None),
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    folder = await _layer_dir(session, layer, project_id=project_id, user_id=user_id)
+    rows = await asyncio.to_thread(layers.list_templates, folder)
+    manifest = await asyncio.to_thread(layers.read_manifest, folder)
+    for row in rows:
+        row["출처"] = manifest.get(f"{row['종류']}/{row['이름']}")
+    return {"층": layer, "양식": rows}
+
+
+@router.get("/layers/{layer}/templates/{kind}/{name}")
+async def read_layer_template(
+    layer: Layer,
+    kind: str,
+    name: str,
+    project_id: str | None = Query(None),
+    user_id: int | None = Query(None),
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    folder = await _layer_dir(session, layer, project_id=project_id, user_id=user_id)
+    try:
+        found = await asyncio.to_thread(layers.read_template, folder, kind, name)
+    except ValueError as error:
+        raise _bad(error) from error
+    if found is None and layer == "project":
+        # 옛 프로젝트(사본 없음) — 시스템 양식으로 떨어짐 · 저장하면 그때 작업본이 생김
+        found = await asyncio.to_thread(layers.read_template, layers.system_dir(), kind, name)
+        if found is not None:
+            found.update({"층": "system", "판": None})
+            return found
+    if found is None:
+        raise HTTPException(status_code=404, detail="양식을 찾을 수 없습니다.")
+    found["층"] = layer
+    found["출처"] = layers.read_manifest(folder).get(f"{kind}/{name}")
+    return found
+
+
+@router.put("/layers/{layer}/templates/{kind}/{name}")
+async def save_layer_template(
+    layer: Layer,
+    kind: str,
+    name: str,
+    body: SaveBody,
+    project_id: str | None = Query(None),
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    folder = await _layer_dir(session, layer, project_id=project_id, write=True)
+    try:
+        version = await asyncio.to_thread(
+            layers.write_template,
+            folder,
+            kind,
+            name,
+            body.문서,
+            version=body.판,
+            check_version=True,
+        )
+    except layers.StaleTemplate as error:
+        raise HTTPException(
+            status_code=409, detail={"message": str(error), "판": error.current}
+        ) from error
+    except ValueError as error:
+        raise _bad(error) from error
+    return {"종류": kind, "이름": name, "판": version, "층": layer}
+
+
+# ── 프로젝트 길 ───────────────────────────────────────
+
+
+@router.post("/projects/{project_id}/templates/reset")
+async def reset_project_templates(
+    project_id: str,
+    body: TargetBody | None = None,
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    project = await _project(session, project_id)
+    target = body or TargetBody()
+    try:
+        done = await asyncio.to_thread(
+            layers.reset_project, project["root"], kind=target.종류, name=target.이름
+        )
+    except ValueError as error:
+        raise _bad(error) from error
+    return {"초기화": done}
+
+
+@router.post("/projects/{project_id}/templates/apply")
+async def apply_project_templates(
+    project_id: str,
+    body: ApplyBody,
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    """다른 층 양식을 작업본에 덮어씀 — [회사 양식 적용] · [양식 가져오기].
+
+    `_initial/` 은 그대로 — 초기화 기준은 안 바뀜.
+    """
+    project = await _project(session, project_id)
+    ref: dict[str, Any] = {}
+    if body.from_ == "project":
+        if not body.project_id or body.project_id == project_id:
+            raise HTTPException(status_code=400, detail="가져올 다른 프로젝트를 고르세요.")
+        other = await _project(session, body.project_id)
+        if other["company_id"] != project["company_id"]:
+            raise HTTPException(status_code=403, detail="같은 회사 프로젝트만 가져옵니다.")
+        source = layers.project_dir(other["root"])
+        ref = {"project_id": body.project_id, "프로젝트": other.get("name")}
+    elif body.from_ == "personal":
+        owner = body.user_id if body.user_id is not None else int(session["user_id"])
+        await _same_company_user(session, owner, int(project["company_id"]))
+        source = layers.personal_dir(project["company_id"], owner)
+        ref = {"user_id": owner}
+    elif body.from_ == "company":
+        source = layers.company_dir(project["company_id"])
+    else:
+        source = layers.system_dir()
+    try:
+        done = await asyncio.to_thread(
+            layers.copy_templates,
+            source,
+            layers.project_dir(project["root"]),
+            source_layer=body.from_,
+            kind=body.종류,
+            name=body.이름,
+            source_ref=ref,
+        )
+    except ValueError as error:
+        raise _bad(error) from error
+    if not done:
+        raise HTTPException(status_code=404, detail="가져올 양식이 없습니다.")
+    return {"적용": done, "from": body.from_}
+
+
+@router.post("/projects/{project_id}/templates/save-as")
+async def save_project_template_as(
+    project_id: str,
+    body: SaveAsBody,
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    """작업본 한 벌을 [내 양식으로 저장] · [회사 공식으로 저장]."""
+    project = await _project(session, project_id)
+    if body.to == "company" and not _is_company_admin(session):
+        raise HTTPException(status_code=403, detail="회사 공식 양식은 회사 관리자만.")
+    if body.to == "company":
+        target = layers.company_dir(project["company_id"])
+    else:
+        target = layers.personal_dir(project["company_id"], session["user_id"])
+    try:
+        done = await asyncio.to_thread(
+            layers.copy_templates,
+            layers.project_dir(project["root"]),
+            target,
+            source_layer="project",
+            kind=body.종류,
+            name=body.이름,
+            source_ref={"project_id": project_id, "프로젝트": project.get("name")},
+        )
+    except ValueError as error:
+        raise _bad(error) from error
+    if not done:
+        raise HTTPException(status_code=404, detail="프로젝트 작업본에 그 양식이 없습니다.")
+    return {"저장": done, "to": body.to}
+
+
+@router.get("/projects/{project_id}/sources")
+async def list_sources(
+    project_id: str,
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    """가져올 수 있는 것 — 시스템 · 회사 공식 · 같은 회사 사람 개인 · 같은 회사 다른 프로젝트."""
+    project = await _project(session, project_id)
+    company_id = int(project["company_id"])
+
+    def _people(users: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        rows = []
+        for user in users:
+            found = layers.list_templates(layers.personal_dir(company_id, user["user_id"]))
+            if found:
+                rows.append({**user, "양식": found})
+        return rows
+
+    def _projects(projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        rows = []
+        for other in projects:
+            if other["project_id"] == str(project_id) or not other.get("storage_path"):
+                continue
+            try:
+                root = Path(resolve_stored_project_path(other["storage_path"]))
+            except ValueError:
+                continue
+            found = layers.list_templates(layers.project_dir(root))
+            if found:
+                rows.append(
+                    {"project_id": other["project_id"], "name": other["name"], "양식": found}
+                )
+        return rows
+
+    users = await _company_users(company_id)
+    projects = await _company_projects(company_id)
+    return {
+        "system": await asyncio.to_thread(layers.list_templates, layers.system_dir()),
+        "company": await asyncio.to_thread(layers.list_templates, layers.company_dir(company_id)),
+        "personal": await asyncio.to_thread(_people, users),
+        "project": await asyncio.to_thread(_projects, projects),
+    }
+
+
+@router.get("/projects/{project_id}/tables/{name}/filled")
+async def filled_table(
+    project_id: str,
+    name: str,
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    """설계값을 채운 표 문서 — 저장 안 함(5장 ③ 서버 단독)."""
+    from M02_MasterTemplete import M02_Table_Fill as fill
+
+    project = await _project(session, project_id)
+    try:
+        found = await asyncio.to_thread(
+            layers.read_template, layers.project_dir(project["root"]), "table", name
+        )
+        if found is None:
+            found = await asyncio.to_thread(
+                layers.read_template, layers.system_dir(), "table", name
+            )
+    except ValueError as error:
+        raise _bad(error) from error
+    if found is None:
+        raise HTTPException(status_code=404, detail="양식을 찾을 수 없습니다.")
+    document = await asyncio.to_thread(fill.fill_table, project["root"], found["문서"])
+    return {"이름": name, "판": found["판"], "문서": document}