eomsangdonandClaude Opus 5.5
3ded39875f
- common_util_spreadsheet_node.ts areaTable: 시트 · 범위 → CAD 표 객체(표시 글 = formatValue · cellStyleAt · 병합 → colSpan/rowSpan · 범위 밖 병합 잘라냄 · 숨긴 행열 뺌 · 가로 정렬 · 굵게 · 열 너비 · 행 높이 비율 · 칸 수 한도 · 칸 key 「시트id!A1」 · style.연결 빈 글)
- 파이썬은 부르기만(common_util_spreadsheet.area_tables · 새 M02_Area_Table.py)
- POST /api/m02/layers/{층}/area/{열}/table {시트, 범위, 글자크기} → {표, 시트, 범위, 값} · 저장 안 함 · 층 읽기 권한
- 면적 저장(Store.write · 비우기 · 복사 · 프로젝트 층 길) 뒤 그 열 도면 연결표마다 표 칸 글 다시 · 마지막 값 갱신 · 표 없으면 기록 둠
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EYJJRCfHfdCGNhmXTjQKV7
378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""M02 층 집계표 API — `/api/m02/layers/{layer}/…` · 항목 목록 · 항목 복사(PLAN 21-1 · 21-2).
|
|
|
|
⚠ 등록은 `main.py`(로그인만) · 층 권한은 `M02_Layer_Access`(양식 길 `Router_Layers` 와 한 벌 ·
|
|
PLAN 27-1c) · `owner` = 회사 `{C}` · 개인 `{C}/{U}`(없으면 자기 층) · 회사 없음 = 409.
|
|
옛 길(`/api/m02/structures/…` 등 · `M02_MasterTemplete_Router.py`)은 시스템 층 별칭으로 그대로.
|
|
회사 · 개인 층은 처음 부를 때 빈 뼈대를 만듦(`Items.ensure`).
|
|
일위대가 갈라 내기(PLAN 27-2c): 새 조합은 그 층에(마스터 UA · 회사 UC · 개인 UP) · 층 쓰기 권한 ·
|
|
첫 진입 뒤 구조물집계표가 이미 있어 같은 이름 새로 쓰기(판 없이)는 409 그대로.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Callable, Coroutine
|
|
from contextvars import ContextVar
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
import anyio.from_thread
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
|
|
from fastapi.routing import APIRoute
|
|
from pydantic import BaseModel
|
|
|
|
from common_util.common_util_audit import record_audit
|
|
from common_util.common_util_auth import verify_session
|
|
from config.config_db import get_db_pool
|
|
from M02_MasterTemplete import M02_Item_Paste as paste_store
|
|
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_MasterTemplete_Tree as tree
|
|
|
|
logger = logging.getLogger(__name__)
|
|
#: 이 요청이 쓴 남의 층(`others_write`) — 길이 성공해 답한 뒤에만 시스템 로그(`NotedRoute`)
|
|
_WRITES: ContextVar[list[tuple[dict, access.Grant]] | None] = ContextVar("m02_writes", default=None)
|
|
|
|
|
|
class NotedRoute(APIRoute):
|
|
"""남의 층 쓰기 기록 = 성공한 쓰기만(39-1d) — 막히거나 실패한 부름(예외 · 4xx · 5xx)은 뺌."""
|
|
|
|
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
|
|
handler = super().get_route_handler()
|
|
|
|
async def run(request: Request) -> Response:
|
|
notes: list[tuple[dict, access.Grant]] = []
|
|
token = _WRITES.set(notes)
|
|
try:
|
|
response = await handler(request)
|
|
finally:
|
|
_WRITES.reset(token)
|
|
if response.status_code < 400:
|
|
for session, got in notes:
|
|
await _note(session, got)
|
|
return response
|
|
|
|
return run
|
|
|
|
|
|
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"], route_class=NotedRoute)
|
|
|
|
Layer = Literal["system", "company", "personal"]
|
|
Session = dict[str, Any]
|
|
NO_COMPANY = access.NO_COMPANY
|
|
|
|
|
|
class TreeBody(BaseModel):
|
|
"""26-1 나무 고침 — `{op, 경로, 이름?, 새경로?, 키?, 판}` (+ 항목 더하기 때 `단위?`)."""
|
|
|
|
op: str
|
|
경로: list[str] | None = None
|
|
이름: str | None = None
|
|
새경로: list[str] | None = None
|
|
키: str | None = None
|
|
단위: str | None = None
|
|
판: str
|
|
# 여러 개 옮기기 · 지우기(PLAN 30-2) — `[{경로, 키?}]` · 있으면 이것마다 차례로 · 저장 한 번
|
|
대상: list[dict[str, Any]] | None = None
|
|
|
|
|
|
class PasteBody(BaseModel):
|
|
"""30-2 층 사이 붙여넣기 — 복사한 층(`from_layer` · `from_owner`) `대상` → 이 층 `경로` 아래."""
|
|
|
|
from_layer: Layer
|
|
from_owner: str | None = None
|
|
대상: list[dict[str, Any]]
|
|
경로: list[str] = []
|
|
|
|
|
|
class CopyItemBody(BaseModel):
|
|
from_layer: Literal["system", "company"]
|
|
from_owner: str | None = None
|
|
key: str
|
|
|
|
|
|
async def _member(owner: str) -> bool:
|
|
"""층 주인이 있나 — 회사 `{C}` = 그 회사 · 개인 `{C}/{U}` = U 가 C 소속."""
|
|
company, _, user = owner.partition("/")
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
if user:
|
|
await cursor.execute(
|
|
"SELECT 1 FROM users WHERE id = %s AND company_id = %s AND deleted_at IS NULL",
|
|
(int(user), int(company)),
|
|
)
|
|
else:
|
|
await cursor.execute(
|
|
"SELECT 1 FROM companies WHERE id = %s AND deleted_at IS NULL", (int(company),)
|
|
)
|
|
return await cursor.fetchone() is not None
|
|
|
|
|
|
async def _note(session: Session, got: access.Grant) -> None:
|
|
"""남의 층 쓰기 한 줄(39-6 · 시스템 로그) — 기록이 못 돼도 쓰기는 막지 않음."""
|
|
try:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await record_audit(
|
|
cursor,
|
|
actor_id=int(session["user_id"]),
|
|
action="M02_OTHER_LAYER_WRITE",
|
|
resource_type=f"m02_{got.layer}",
|
|
resource_ref=got.owner,
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
logger.exception("M02 남의 층 쓰기 기록 실패: %s %s", got.layer, got.owner)
|
|
|
|
|
|
async def others_write(session: Session, got: access.Grant, folder: Path) -> None:
|
|
"""남의 층 쓰기 — 처음(집계표 없음)이면 주인이 있는지 본 뒤 뼈대 · 시스템 로그는 길이 성공한
|
|
뒤(`NotedRoute` · 그 밖의 길은 바로)."""
|
|
if not (folder / "table" / f"{store.STRUCTURE_TABLE}.json").is_file():
|
|
if not await _member(str(got.owner)):
|
|
raise HTTPException(status_code=403, detail="그 회사 · 그 회사 사람이 아님")
|
|
items_store.ensure(folder)
|
|
pending = _WRITES.get()
|
|
if pending is None:
|
|
await _note(session, got)
|
|
else:
|
|
pending.append((session, got))
|
|
|
|
|
|
def _root(session: Session, layer: str, owner: str | None = None, write: bool = False) -> Path:
|
|
"""층 폴더(권한까지 · `M02_Layer_Access`) — 자기 회사 · 개인 층은 없으면 빈 뼈대를 만듦 ·
|
|
남의 층 쓰기(맡은 회사 · 39-6)는 `others_write`."""
|
|
got = access.require(session, layer, owner, write=write)
|
|
if layer == "system":
|
|
return store.FOLDER
|
|
folder = access.folder(got, create=write)
|
|
if got.mine:
|
|
items_store.ensure(folder)
|
|
elif write:
|
|
anyio.from_thread.run(others_write, session, got, folder)
|
|
elif not (folder / "table" / f"{store.STRUCTURE_TABLE}.json").is_file():
|
|
raise HTTPException(status_code=404, detail="그 층은 아직 만든 것이 없음")
|
|
return folder
|
|
|
|
|
|
def _call(folder: Path, fn, *args):
|
|
with store.at(folder):
|
|
try:
|
|
return fn(*args)
|
|
except store.StoreError as e:
|
|
raise HTTPException(status_code=e.status, detail=e.detail) from e
|
|
|
|
|
|
# ── 항목 목록 · 복사 ───────────────────────────────────
|
|
|
|
|
|
def _listing(session: Session, layer: str, owner: str | None) -> dict[str, Any]:
|
|
if layer != "system" and not owner and access.own_owner(session, layer) is None:
|
|
return {"layer": layer, "owner": None, "items": []} # 회사 없음 = 빈 그룹
|
|
who = access.grant(session, layer, owner).owner
|
|
folder = _root(session, layer, who)
|
|
return {"layer": layer, "owner": who, "items": items_store.items(folder)}
|
|
|
|
|
|
@router.get("/items")
|
|
def get_items(
|
|
layer: Layer = "system",
|
|
owner: str | None = None,
|
|
all: bool = False, # noqa: A002 — 21-1 약속 이름 `?all=1`
|
|
session: Session = Depends(verify_session),
|
|
):
|
|
"""21-1 — 한 층이면 `[{key, 열, 종류, 분류1~3, 단위, 일위대가, 도번, 면적번}]` ·
|
|
`?all=1` 이면 세 층 `[{layer, owner, items}]`(회사 · 개인은 세션 것 · 회사 없으면 빈 목록)."""
|
|
if all:
|
|
return [_listing(session, one, None) for one in ("system", "company", "personal")]
|
|
return _listing(session, layer, owner)["items"]
|
|
|
|
|
|
@router.post("/layers/personal/items/copy")
|
|
def copy_item(body: CopyItemBody, session: Session = Depends(verify_session)) -> dict:
|
|
"""회사 · 마스터 항목 → 내 개인 층 새 항목(열 · 세 파일 · 새 SS 키 · `본뜬`)."""
|
|
target = _root(session, "personal", None, write=True)
|
|
source = _root(session, body.from_layer, body.from_owner)
|
|
owner = access.grant(session, body.from_layer, body.from_owner).owner
|
|
origin = {"layer": body.from_layer, "owner": owner, "key": body.key}
|
|
try:
|
|
return items_store.copy_item(source, target, body.key, origin)
|
|
except store.StoreError as e:
|
|
raise HTTPException(status_code=e.status, detail=e.detail) from e
|
|
|
|
|
|
# ── 폴더 구조(PLAN 26-1 · 26-2a) — 읽기 = 층 읽기 권한 · 고침 = 층 쓰기 권한 ──
|
|
|
|
|
|
def _tree_view(folder: Path) -> dict[str, Any]:
|
|
"""나무 — `{판(집계표), 폴더: [경로 …], items: [항목 …]}` · 폴더 = 빈 폴더 + 항목 앞마디."""
|
|
with store.at(folder):
|
|
got = store.read("table", store.STRUCTURE_TABLE)
|
|
doc = tree.normalize_table(got["문서"]) # 옛 파일도 경로로 읽음
|
|
return {"판": got["판"], "폴더": tree.folders_of(doc), "items": items_store.items(folder)}
|
|
|
|
|
|
@router.get("/layers/{layer}/tree")
|
|
def get_tree(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
|
|
folder = _root(session, layer, owner)
|
|
return _call(folder, _tree_view, folder)
|
|
|
|
|
|
def _edit_tree(folder: Path, body: dict[str, Any]) -> dict[str, Any]:
|
|
"""집계표를 고쳐 집계표 저장 한 줄기로 — 번호 · SS 장부 · 지운 항목의 세 파일 · 입력 연결."""
|
|
got = store.read("table", store.STRUCTURE_TABLE)
|
|
if got["판"] != body.get("판"):
|
|
raise store.StoreError(409, {"stale": [store.STRUCTURE_TABLE], "판": got["판"]})
|
|
ids = {} # 키 · 옛 키(회사 · 개인 층 SS) → 열 id
|
|
for r in store.read_combos().get("줄", []):
|
|
for key in [r.get("키"), *(r.get("옛키") or [])]:
|
|
ids[str(key)] = str(r.get("열"))
|
|
try:
|
|
targets = body.get("대상")
|
|
if targets is None:
|
|
doc, note = tree.apply(got["문서"], body, ids)
|
|
else: # 여러 개 — 차례로 고쳐 저장은 한 번(하나라도 막히면 아무것도 안 씀)
|
|
if body.get("op") not in ("move", "delete") or not targets:
|
|
raise tree.TreeError(422, "여러 개는 옮기기 · 지우기만 — 「대상」 이 필요함")
|
|
doc, note = got["문서"], {"op": body["op"], "지운항목": []}
|
|
for one in paste_store.outer(targets):
|
|
step = {**body, "경로": one.get("경로"), "키": one.get("키")}
|
|
doc, done = tree.apply(doc, step, ids)
|
|
note["지운항목"] += done.get("지운항목", [])
|
|
except tree.TreeError as e:
|
|
raise store.StoreError(e.status, e.detail) from e
|
|
store.write("table", store.STRUCTURE_TABLE, got["판"], doc)
|
|
return {**_tree_view(folder), "알림": note}
|
|
|
|
|
|
@router.post("/layers/{layer}/tree")
|
|
def post_tree(
|
|
layer: Layer,
|
|
body: TreeBody,
|
|
owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
):
|
|
"""폴더 더하기 · 항목 더하기 · 이름 바꾸기 · 옮기기 · 지우기 → 새 나무 · 판 다르면 409.
|
|
쓰이는 항목을 지워도 프로젝트엔 영향 없음(프로젝트는 복사본을 씀 · 알림의 `지운항목`)."""
|
|
folder = _root(session, layer, owner, write=True)
|
|
return _call(folder, _edit_tree, folder, body.model_dump())
|
|
|
|
|
|
@router.post("/layers/{layer}/items/paste")
|
|
def post_paste(
|
|
layer: Layer,
|
|
body: PasteBody,
|
|
owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
):
|
|
"""여러 항목 · 폴더를 층 사이 붙여넣기(PLAN 30-2) — 복사한 층은 읽기 권한 · 이 층은 쓰기 권한 ·
|
|
새 키(SC · SP) · 본뜬 = 원본 · 같은 이름 「이름 (2)」 · 폴더는 안의 것까지 → 새 나무 + 알림."""
|
|
target = _root(session, layer, owner, write=True)
|
|
source = _root(session, body.from_layer, body.from_owner)
|
|
who = access.grant(session, body.from_layer, body.from_owner).owner
|
|
try:
|
|
note = paste_store.paste(
|
|
source, target, body.대상, body.경로, {"layer": body.from_layer, "owner": who}
|
|
)
|
|
except store.StoreError as e:
|
|
raise HTTPException(status_code=e.status, detail=e.detail) from e
|
|
return {**_call(target, _tree_view, target), "알림": note}
|
|
|
|
|
|
# ── 층 길 — 옛 시스템 길과 같은 모양 ─────────────────────
|
|
|
|
|
|
@router.get("/layers/{layer}/structures/numbers")
|
|
def get_numbers(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
|
|
return _call(_root(session, layer, owner), store.structure_numbers)
|
|
|
|
|
|
@router.get("/layers/system/structure-options")
|
|
def get_structure_options(session: Session = Depends(verify_session)) -> dict:
|
|
"""구조물 옵션 보이기 읽기 — 로그인 누구나(마스터 한 벌 · 고쳐 씀은 옛 길 시스템 관리자만)."""
|
|
from M02_MasterTemplete import M02_MasterTemplete_Options as options
|
|
|
|
return _call(_root(session, "system"), options.listing)
|
|
|
|
|
|
@router.get("/layers/{layer}/structure-combos")
|
|
def get_combos(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
|
|
return _call(_root(session, layer, owner), store.read_combos)
|
|
|
|
|
|
@router.post("/layers/{layer}/structures/{column}")
|
|
def post_structure(
|
|
layer: Layer, column: str, owner: str | None = None, session: Session = Depends(verify_session)
|
|
):
|
|
return _call(_root(session, layer, owner, write=True), store.create_structure, column)
|
|
|
|
|
|
def _copy_practice(column: str, key: str) -> dict[str, Any]:
|
|
"""실무문서 조합(마스터 UA) → 이 열 일위대가 문서(입력값 빈 값 · 39-1b)."""
|
|
from M02_MasterTemplete import M02_Labor_Doc as labor
|
|
|
|
if not key.startswith("UA"):
|
|
raise store.StoreError(400, f"실무문서 조합 키가 아님 「{key}」")
|
|
return store.copy_from("labor", column, key, body=labor.from_combo(key))
|
|
|
|
|
|
class AreaTableBody(BaseModel):
|
|
"""면적산출 범위 → 도면 표(41-1) — `시트` = 시트 id · 이름 · `글자크기` 기본 2.5."""
|
|
|
|
시트: str
|
|
범위: str
|
|
글자크기: float | None = None
|
|
|
|
|
|
@router.post("/layers/{layer}/area/{column}/table")
|
|
def post_area_table(
|
|
layer: Layer,
|
|
column: str,
|
|
body: AreaTableBody,
|
|
owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
) -> dict:
|
|
"""그 열 면적산출(저장본) 범위 → CAD 표 `{표, 시트, 범위, 값}` — 저장 안 함 · 층 읽기 권한."""
|
|
from M02_MasterTemplete import M02_Area_Table as area_table
|
|
|
|
folder = _root(session, layer, owner)
|
|
return _call(folder, area_table.to_table, column, body.시트, body.범위, body.글자크기)
|
|
|
|
|
|
@router.post("/layers/{layer}/{kind}/{column}/clear")
|
|
def post_clear(
|
|
layer: Layer,
|
|
kind: Literal["structures", "labor", "area"],
|
|
column: str,
|
|
owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
):
|
|
"""항목 하나 비우기(40-4 · 카드 [삭제]) — 빈 문서 · 번호 · 열 그대로 · 받는 층 쓰기 권한."""
|
|
folder = _root(session, layer, owner, write=True)
|
|
part = "structure" if kind == "structures" else kind
|
|
return _call(folder, store.clear, part, column)
|
|
|
|
|
|
@router.post("/layers/{layer}/{kind}/{column}/copy")
|
|
def post_copy(
|
|
layer: Layer,
|
|
kind: Literal["structures", "labor", "area"],
|
|
column: str,
|
|
source: str = Query(alias="from"),
|
|
owner: str | None = None,
|
|
from_layer: Literal["system", "company", "personal", "practice"] | None = None,
|
|
from_owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
):
|
|
"""항목 하나 복사해 가져오기(덮어씀 · 번호 · 열 그대로) — 원본 = 같은 층(`from_layer` 없음) ·
|
|
다른 층(원본 읽기 권한) · 실무문서 조합(`practice` · 일위대가만 · 시스템 관리자만)."""
|
|
folder = _root(session, layer, owner, write=True)
|
|
part = "structure" if kind == "structures" else kind
|
|
if from_layer == "practice":
|
|
if part != "labor" or not access.is_system_admin(session):
|
|
raise HTTPException(status_code=403, detail="실무문서는 시스템 관리자가 일위대가로만")
|
|
return _call(folder, _copy_practice, column, source)
|
|
origin = _root(session, from_layer, from_owner) if from_layer else None
|
|
return _call(folder, store.copy_from, part, column, source, origin)
|