Aislo/M02_MasterTemplete/M02_MasterTemplete_Router_Layers.py

"""M02 양식 층 API — 계약 `tmp/M02_분석/6_계약.md` 「층 (sub4)」.

⚠ 등록은 `main.py`(로그인만) · 시스템 · 회사 · 개인 층 권한은 `M02_Layer_Access`(나무 · 항목 길과
한 벌 · PLAN 27-1c · `owner` = 회사 `{C}` · 개인 `{C}/{U}`) · 남의 층 목록 = `/layers/{층}/owners`:
    project   같은 회사 프로젝트 · 작업본 쓰기 · `_initial/` 은 안 씀 ·
              가져오기(apply · sources)는 같은 회사 사람 개인 층도 읽음
"""

from __future__ import annotations

import asyncio
import logging
from pathlib import Path
from typing import Any, Literal
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, ConfigDict, Field

from B06_Section.B06_Section_Repository import (
    get_cross_section_designs,
    get_workflow_route_context,
)
from common_util import common_util_spreadsheet as spreadsheet
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, run_with_connection
from M02_MasterTemplete import M02_Layer_Access as access
from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Table_Fill as fill
from M02_MasterTemplete import M02_Template_Layers as layers

logger = logging.getLogger(__name__)

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]


async def _companies() -> 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 companies WHERE deleted_at IS NULL ORDER BY id")
        rows = await cursor.fetchall()
    return [{"company_id": row[0], "name": row[1]} for row in rows]


async def _cross_designs(project_id: str) -> list[dict[str, Any]]:
    """최신 노선의 B06 횡단 설계 — 못 읽으면 빈 목록(관 연장이 빈칸으로 섬 · 0 아님)."""
    try:
        context = await run_with_connection(get_workflow_route_context, UUID(str(project_id)))
        route_id = int((context or {}).get("route_id") or 0)
        return await run_with_connection(get_cross_section_designs, route_id) if route_id else []
    except Exception:
        logger.exception("M02 채운 표 — 횡단 설계 조회 실패: project_id=%s", project_id)
        return []


# ── 권한 · 자리 ───────────────────────────────────────


_is_system_admin = access.is_system_admin
_is_company_admin = access.is_company_admin


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:  # 회사 없는 사용자는 없음(21-0) — 회사 · 개인 층은 회사부터
        raise HTTPException(status_code=409, 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,
    owner: str | None = None,
    write: bool = False,
) -> Path:
    """층 폴더 — 권한은 `M02_Layer_Access` 한 벌(나무 · 항목 길과 같음 · PLAN 27-1c) ·
    `owner` 없으면 프로젝트(없으면 세션) 회사의 회사 층 · 개인 층(`user_id` 또는 본인)."""
    if layer == "project":
        return layers.project_dir((await _project(session, project_id))["root"])
    if layer == "system":
        access.require(session, layer, write=write)
        return layers.system_dir()
    if not owner:
        project = await _project(session, project_id) if project_id else None
        company_id = _company_of(session, project)
        who = session["user_id"] if user_id is None else user_id
        owner = str(company_id) if layer == "company" else f"{company_id}/{who}"
    got = access.require(session, layer, owner, write=write)
    return _ensured(access.folder(got)) if got.mine else access.folder(got)


def _ensured(folder: Path) -> Path:
    """회사 · 개인 층 첫 진입 — 빈 구조물집계표 뼈대(PLAN 21-2 · `Items.ensure`)."""
    items_store.ensure(folder)
    return folder


def _bad(error: ValueError) -> HTTPException:
    return HTTPException(status_code=400, detail=str(error))


# ── 층 길 ─────────────────────────────────────────────


@router.get("/layers/{layer}/owners")
async def list_owners(
    layer: Literal["company", "personal"],
    company: int | None = Query(None),
    session: dict[str, Any] = Depends(verify_session),
) -> list[dict[str, Any]]:
    """남의 층 목록(PLAN 27-1c · 펼칠 때 부름) — `[{owner, name}]` · 층 자리가 있는 것만(본인 뺌).
    회사 = SYSTEM_ADMIN 만 회사 목록 · 개인 = 그 회사 사람(SYSTEM_ADMIN 은 어느 회사나 ·
    회사 관리자는 자기 회사만)."""
    if layer == "company":
        if not access.is_system_admin(session):
            raise HTTPException(status_code=403, detail="회사 목록은 시스템 관리자만 봄.")
        rows = [(str(r["company_id"]), r["name"]) for r in await _companies()]
    else:
        target = session.get("company_id") if company is None else company
        if target is None:
            raise HTTPException(status_code=409, detail=access.NO_COMPANY)
        mine = str(target) == str(session.get("company_id"))
        if not (access.is_system_admin(session) or (mine and access.is_company_admin(session))):
            raise HTTPException(status_code=403, detail="같은 회사 사람 목록은 회사 관리자만 봄.")
        me = session.get("user_id")
        users = await _company_users(int(target))
        rows = [(f"{target}/{u['user_id']}", u["name"]) for u in users if u["user_id"] != me]
    found = []
    for owner, name in rows:
        got = access.Grant(layer, owner, False, owner == access.own_owner(session, layer))
        try:
            access.folder(got)
        except HTTPException:
            continue
        found.append({"owner": owner, "name": name})
    return found


@router.get("/layers/{layer}/templates")
async def list_layer(
    layer: Layer,
    project_id: str | None = Query(None),
    user_id: int | None = Query(None),
    owner: str | 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, owner=owner)
    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),
    owner: str | 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, owner=owner)
    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),
    user_id: int | None = Query(None),
    owner: str | 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, owner=owner, write=True
    )
    document = body.문서
    try:
        layers.check_skeleton(kind, document)  # 빈 `{}` · 뼈대 없는 문서는 파일 안 씀
    except ValueError as error:
        raise _bad(error) from error
    if kind == "table" and fill.is_fillable(document):
        # 채운 표가 와도 설계값은 안 받아 적음 — 양식 + 손 값만(5장 · 브라우저 값을 믿지 않음)
        document = fill.strip_design(document)
    if layer != "project":
        # 시스템 · 회사 · 개인 층 = 마스터 저장소 규칙 그대로(집계표 저장 때 새 열 파일 ·
        # 번호 · SS 장부 ·
        # 계산값 서버 풀이 · 번호 칸 보존 — `M02_MasterTemplete_Store.write`)
        return await asyncio.to_thread(
            _store_save, folder, kind, name, body.판 or "", document, layer
        )
    if kind in ("area", "material"):
        # 면적산출 · 재료산출 계산값은 서버가 같은 TS 엔진을 Node 로 돌려 새로 적음
        # (브라우저 값 버림 · 5장 ②)
        try:
            document = await asyncio.to_thread(spreadsheet.with_server_values, document)
        except spreadsheet.RecalcError as error:
            raise HTTPException(status_code=503, detail=str(error)) from error
    try:
        version = await asyncio.to_thread(
            layers.write_template,
            folder,
            kind,
            name,
            document,
            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}


def _store_save(folder: Path, kind: str, name: str, version: str, doc: Any, layer: str) -> dict:
    with store.at(folder):
        try:
            info = store.write(kind, name, version, doc, screen=True)  # 화면 [저장](26-7e)
        except store.StoreError as error:
            raise HTTPException(status_code=error.status, detail=error.detail) from error
    return {"종류": kind, "이름": name, "판": info["판"], "층": layer}


@router.delete("/layers/{layer}/templates/{kind}/{name}")
async def delete_layer_template(
    layer: Layer,
    kind: str,
    name: str,
    판: str | None = Query(None),
    project_id: str | None = Query(None),
    owner: str | None = Query(None),
    session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
    """개인 = 본인만 · 회사 = 회사 관리자만 · 시스템 = 시스템 관리자 · 프로젝트 층은 거절 ·
    판이 다르면 409 · 프로그램이 이름으로 부르는 양식(`store.IN_USE` — 구조물집계표 등)은 409."""
    if layer == "project":
        raise HTTPException(
            status_code=403, detail="개인 · 회사 · 시스템 층 양식만 지울 수 있습니다."
        )
    folder = await _layer_dir(session, layer, project_id=project_id, owner=owner, write=True)
    if (kind, name) in store.IN_USE:
        raise HTTPException(status_code=409, detail=f"쓰는 양식이라 못 지움 「{name}」")
    try:
        layers.check_kind(kind)
        layers.check_name(name)
    except ValueError as error:
        raise _bad(error) from error
    if not 판:
        raise HTTPException(status_code=400, detail="지울 양식의 판이 필요합니다.")
    try:
        found = await asyncio.to_thread(
            layers.delete_template, folder, kind, name, version=판, check_version=True
        )
    except layers.StaleTemplate as error:
        raise HTTPException(
            status_code=409, detail={"message": str(error), "판": error.current}
        ) from error
    if not found:
        raise HTTPException(status_code=404, detail="양식을 찾을 수 없습니다.")
    return {"종류": kind, "이름": name, "층": layer, "지움": True}


# ── 프로젝트 길 ───────────────────────────────────────


@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장 ③ 서버 단독) · `결과` = 계산 열(없으면 null)."""
    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="양식을 찾을 수 없습니다.")
    lengths = fill.pipe_lengths_from_designs(await _cross_designs(project_id))
    document = await asyncio.to_thread(fill.fill_table, project["root"], found["문서"], lengths)
    result = await asyncio.to_thread(fill.recalc, document)
    return {"이름": name, "판": found["판"], "문서": document, "결과": result}