Merge remote-tracking branch 'origin/sub_laptop_4' into sub_laptop_2
This commit is contained in:
@@ -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}
|
||||
@@ -0,0 +1,312 @@
|
||||
"""M02 양식 층 — 네 층의 자리 · 읽기 · 쓰기 · 복사 · manifest.
|
||||
|
||||
층 넷 (PLAN 10-5):
|
||||
system `resources/master_template/{table,drawing}/<이름>.json` (git)
|
||||
company `storage/{회사}/templates/{table,drawing}/`
|
||||
personal `storage/{회사}/{사용자}/templates/{table,drawing}/`
|
||||
project `storage/{회사}/{사용자}/{프로젝트}/templates/{table,drawing}/`
|
||||
+ 초기 사본 `templates/_initial/`
|
||||
|
||||
- 층 폴더마다 `manifest.json` — 그 폴더에 든 양식의 출처 `{"table/이름": {층, 이름, 판, 적용일}}`.
|
||||
- 판 = 파일 sha256 앞 16자(M01 Store 와 같음) · 판이 다르면 `StaleTemplate`.
|
||||
- `_initial/` 은 프로젝트 첫 복사 때만 씀 — 그 뒤 어떤 길로도 쓰지 않음(초기화는 읽기만).
|
||||
- 권한은 부르는 쪽(라우터) 몫 — 여기는 자리와 파일만.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from config import config_system
|
||||
|
||||
LAYERS = ("system", "company", "personal", "project")
|
||||
KINDS = ("table", "drawing")
|
||||
TEMPLATES_DIRNAME = "templates"
|
||||
INITIAL_DIRNAME = "_initial"
|
||||
MANIFEST_NAME = "manifest.json"
|
||||
|
||||
#: 시스템 층 뿌리 — 시험이 임시 폴더로 바꿈.
|
||||
SYSTEM_ROOT = config_system.PROJECT_ROOT / "resources" / "master_template"
|
||||
|
||||
|
||||
class StaleTemplate(Exception):
|
||||
"""저장하려는 판이 파일의 지금 판과 다름 — 라우터가 409 로 답함."""
|
||||
|
||||
def __init__(self, current: str | None) -> None:
|
||||
super().__init__("양식이 그새 바뀌었습니다.")
|
||||
self.current = current
|
||||
|
||||
|
||||
# ── 자리 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def storage_root() -> Path:
|
||||
"""`storage/` 실경로 — 부를 때마다 설정을 읽음(시험이 바꿈)."""
|
||||
return Path(config_system.STORAGE_BASE_DIR).resolve()
|
||||
|
||||
|
||||
def _inside(root: Path, path: Path) -> Path:
|
||||
path = path.resolve()
|
||||
if root.resolve() not in (path, *path.parents):
|
||||
raise ValueError("양식 자리가 저장소 루트를 벗어났습니다.")
|
||||
return path
|
||||
|
||||
|
||||
def _id(value: Any, what: str) -> str:
|
||||
text = str(value)
|
||||
if not text.isdigit():
|
||||
raise ValueError(f"{what} 식별자가 올바르지 않습니다.")
|
||||
return text
|
||||
|
||||
|
||||
def system_dir() -> Path:
|
||||
return Path(SYSTEM_ROOT)
|
||||
|
||||
|
||||
def company_dir(company_id: int | str) -> Path:
|
||||
root = storage_root()
|
||||
return _inside(root, root / _id(company_id, "회사") / TEMPLATES_DIRNAME)
|
||||
|
||||
|
||||
def personal_dir(company_id: int | str, user_id: int | str) -> Path:
|
||||
root = storage_root()
|
||||
return _inside(
|
||||
root, root / _id(company_id, "회사") / _id(user_id, "사용자") / TEMPLATES_DIRNAME
|
||||
)
|
||||
|
||||
|
||||
def project_dir(project_root: str | Path) -> Path:
|
||||
"""프로젝트 작업본 자리 — `project_root` 는 `resolve_stored_project_path` 결과."""
|
||||
return Path(project_root) / TEMPLATES_DIRNAME
|
||||
|
||||
|
||||
def initial_dir(project_root: str | Path) -> Path:
|
||||
return project_dir(project_root) / INITIAL_DIRNAME
|
||||
|
||||
|
||||
def check_kind(kind: str) -> str:
|
||||
if kind not in KINDS:
|
||||
raise ValueError(f"양식 종류는 {', '.join(KINDS)} 중 하나입니다.")
|
||||
return kind
|
||||
|
||||
|
||||
def check_name(name: str) -> str:
|
||||
text = str(name or "").strip()
|
||||
bad = not text or text != name or text.startswith((".", "_")) or len(text) > 80
|
||||
if bad or any(ch in text for ch in '/\\:*?"<>|') or ".." in text:
|
||||
raise ValueError("양식 이름이 올바르지 않습니다.")
|
||||
return text
|
||||
|
||||
|
||||
def template_path(layer_dir: str | Path, kind: str, name: str) -> Path:
|
||||
base = Path(layer_dir)
|
||||
return _inside(base, base / check_kind(kind) / f"{check_name(name)}.json")
|
||||
|
||||
|
||||
# ── 읽기 · 쓰기 ───────────────────────────────────────
|
||||
|
||||
|
||||
def version_of(path: str | Path) -> str | None:
|
||||
path = Path(path)
|
||||
if not path.is_file():
|
||||
return None
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()[:16]
|
||||
|
||||
|
||||
def _modified(path: Path) -> str:
|
||||
return datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def list_templates(layer_dir: str | Path) -> list[dict[str, Any]]:
|
||||
"""층 하나의 양식 목록 `[{종류, 이름, 판, 수정일}]` — 종류 · 이름 차례."""
|
||||
base = Path(layer_dir)
|
||||
rows: list[dict[str, Any]] = []
|
||||
for kind in KINDS:
|
||||
folder = base / kind
|
||||
if not folder.is_dir():
|
||||
continue
|
||||
for path in sorted(folder.glob("*.json")):
|
||||
if path.name.startswith((".", "_")):
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"종류": kind,
|
||||
"이름": path.stem,
|
||||
"판": version_of(path),
|
||||
"수정일": _modified(path),
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def read_template(layer_dir: str | Path, kind: str, name: str) -> dict[str, Any] | None:
|
||||
"""`{종류, 이름, 판, 문서}` — 없으면 None."""
|
||||
path = template_path(layer_dir, kind, name)
|
||||
if not path.is_file():
|
||||
return None
|
||||
raw = path.read_bytes()
|
||||
return {
|
||||
"종류": kind,
|
||||
"이름": name,
|
||||
"판": hashlib.sha256(raw).hexdigest()[:16],
|
||||
"문서": json.loads(raw.decode("utf-8")),
|
||||
}
|
||||
|
||||
|
||||
def write_template(
|
||||
layer_dir: str | Path,
|
||||
kind: str,
|
||||
name: str,
|
||||
document: dict[str, Any],
|
||||
*,
|
||||
version: str | None = None,
|
||||
check_version: bool = False,
|
||||
) -> str:
|
||||
"""원자 쓰기 → 새 판. `check_version` 이면 `version` 이 지금 판과 같아야 함(새 파일은 None)."""
|
||||
if not isinstance(document, dict):
|
||||
raise ValueError("양식 문서는 JSON 객체여야 합니다.")
|
||||
path = template_path(layer_dir, kind, name)
|
||||
if check_version:
|
||||
current = version_of(path)
|
||||
if current != (version or None):
|
||||
raise StaleTemplate(current)
|
||||
atomic_write_json(path, document)
|
||||
return version_of(path) or ""
|
||||
|
||||
|
||||
def delete_template(layer_dir: str | Path, kind: str, name: str) -> bool:
|
||||
path = template_path(layer_dir, kind, name)
|
||||
if not path.is_file():
|
||||
return False
|
||||
path.unlink()
|
||||
manifest = read_manifest(layer_dir)
|
||||
if manifest.pop(f"{kind}/{name}", None) is not None:
|
||||
_write_manifest(layer_dir, manifest)
|
||||
return True
|
||||
|
||||
|
||||
# ── manifest ──────────────────────────────────────────
|
||||
|
||||
|
||||
def read_manifest(layer_dir: str | Path) -> dict[str, Any]:
|
||||
path = Path(layer_dir) / MANIFEST_NAME
|
||||
if not path.is_file():
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
return data.get("양식", {}) if isinstance(data, dict) else {}
|
||||
|
||||
|
||||
def _write_manifest(layer_dir: str | Path, entries: dict[str, Any]) -> None:
|
||||
atomic_write_json(Path(layer_dir) / MANIFEST_NAME, {"양식": entries})
|
||||
|
||||
|
||||
def _stamp(layer: str, name: str, version: str | None, **extra: Any) -> dict[str, Any]:
|
||||
entry = {
|
||||
"층": layer,
|
||||
"이름": name,
|
||||
"판": version,
|
||||
"적용일": datetime.now().isoformat(timespec="seconds"),
|
||||
}
|
||||
entry.update({key: value for key, value in extra.items() if value is not None})
|
||||
return entry
|
||||
|
||||
|
||||
# ── 복사 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def copy_templates(
|
||||
source_dir: str | Path,
|
||||
target_dir: str | Path,
|
||||
*,
|
||||
source_layer: str,
|
||||
kind: str | None = None,
|
||||
name: str | None = None,
|
||||
only_missing: bool = False,
|
||||
source_ref: dict[str, Any] | None = None,
|
||||
) -> list[str]:
|
||||
"""`source_dir` 층의 양식을 `target_dir` 로 복사 → 복사한 `종류/이름` 목록.
|
||||
|
||||
`kind`·`name` 으로 좁힘 · `only_missing` 이면 이미 있는 파일은 건너뜀(더하기만).
|
||||
manifest 에 출처(층 · 이름 · 판 · `source_ref`)를 적음. 출처 manifest 의 원래 출처는
|
||||
적지 않음 — 「어디서 가져왔나」 한 단계만.
|
||||
"""
|
||||
if source_layer not in LAYERS:
|
||||
raise ValueError("출처 층이 올바르지 않습니다.")
|
||||
rows = list_templates(source_dir)
|
||||
if kind is not None:
|
||||
rows = [row for row in rows if row["종류"] == check_kind(kind)]
|
||||
if name is not None:
|
||||
rows = [row for row in rows if row["이름"] == check_name(name)]
|
||||
manifest = read_manifest(target_dir)
|
||||
copied: list[str] = []
|
||||
for row in rows:
|
||||
source = template_path(source_dir, row["종류"], row["이름"])
|
||||
target = template_path(target_dir, row["종류"], row["이름"])
|
||||
if only_missing and target.exists():
|
||||
continue
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = target.with_name(f".{target.name}.tmp")
|
||||
shutil.copyfile(source, temporary)
|
||||
temporary.replace(target)
|
||||
key = f"{row['종류']}/{row['이름']}"
|
||||
manifest[key] = _stamp(source_layer, row["이름"], row["판"], **(source_ref or {}))
|
||||
copied.append(key)
|
||||
if copied:
|
||||
_write_manifest(target_dir, manifest)
|
||||
return copied
|
||||
|
||||
|
||||
def seed_project(project_root: str | Path, *, only_missing: bool = False) -> dict[str, list[str]]:
|
||||
"""시스템 양식 전부를 프로젝트 작업본과 `_initial/` 로 복사.
|
||||
|
||||
프로젝트 만들 때 · 옛 프로젝트 넣기(`only_missing`) 둘 다 이 길.
|
||||
|
||||
`_initial/` 은 늘 없는 것만 더함 — 이미 있으면 절대 덮지 않음.
|
||||
"""
|
||||
return {
|
||||
"작업본": copy_templates(
|
||||
system_dir(),
|
||||
project_dir(project_root),
|
||||
source_layer="system",
|
||||
only_missing=only_missing,
|
||||
),
|
||||
"초기": copy_templates(
|
||||
system_dir(), initial_dir(project_root), source_layer="system", only_missing=True
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def reset_project(
|
||||
project_root: str | Path, *, kind: str | None = None, name: str | None = None
|
||||
) -> list[str]:
|
||||
"""[초기화] — `_initial/` 을 작업본에 덮어씀(`_initial/` 은 읽기만)."""
|
||||
initial = initial_dir(project_root)
|
||||
rows = list_templates(initial)
|
||||
if kind is not None:
|
||||
rows = [row for row in rows if row["종류"] == check_kind(kind)]
|
||||
if name is not None:
|
||||
rows = [row for row in rows if row["이름"] == check_name(name)]
|
||||
initial_manifest = read_manifest(initial)
|
||||
manifest = read_manifest(project_dir(project_root))
|
||||
done: list[str] = []
|
||||
for row in rows:
|
||||
key = f"{row['종류']}/{row['이름']}"
|
||||
target = template_path(project_dir(project_root), row["종류"], row["이름"])
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = target.with_name(f".{target.name}.tmp")
|
||||
shutil.copyfile(template_path(initial, row["종류"], row["이름"]), temporary)
|
||||
temporary.replace(target)
|
||||
manifest[key] = initial_manifest.get(key) or _stamp("system", row["이름"], row["판"])
|
||||
done.append(key)
|
||||
if done:
|
||||
_write_manifest(project_dir(project_root), manifest)
|
||||
return done
|
||||
@@ -0,0 +1,228 @@
|
||||
{
|
||||
"양식": "구조물집계표",
|
||||
"종류": "표",
|
||||
"판": 1,
|
||||
"설명": "측점마다 구조물 수량 한 줄 · 머리 4층(종류 · 공법 · 규격 · 단위) · 설계값 열은 프로젝트 설계에서 채움(화면에서 못 고침 — 설계를 고침) · 손 열은 사용자가 적음 · 계산 열은 식. 종류 정본 = B05_Profile/B05_Profile_Structure_Types.json.",
|
||||
"바인딩규칙": "종류 = type_id · 펼침 = 값마다 열을 나눌 칸 키(앞에 = 이면 고정 글) · 값 = 칸 키 | 개소(1건 1) | 길이(length_m, 없으면 끝-시작) | 관연장(B06 횡단 pipe_length_m) · 묶음 = 같은 펼침으로 함께 나뉘는 열 · 머리틀 = 펼친 머리({0}{1}… = 펼침 값 차례) · 이름 = 설계 값 → 머리 글 · 조건 = {칸: 값}(맞는 것만) · 더함 = 같은 열에 더하는 다른 출처 · 식틀 = 펼친 열의 식({0}… 펼침 값)",
|
||||
"층": ["종류", "공법", "규격"],
|
||||
"변수": {
|
||||
"파형강관_1본_m": 8,
|
||||
"흄관_1본_m": "",
|
||||
"VR관_1본_m": "",
|
||||
"수축줄눈_간격_m": 6,
|
||||
"측점간격_m": 20
|
||||
},
|
||||
"쪽줄": 50,
|
||||
"열": [
|
||||
{ "id": "no", "머리": ["NO", null, null], "단위": null, "꼴": "글", "설명": "줄 차례(전구간 줄 빼고 1부터)" },
|
||||
{ "id": "sta", "머리": ["측점", null, null], "단위": null, "꼴": "글", "설명": "점 NO.x+y · 구간 NO.a~NO.b · 맨 위 전구간" },
|
||||
|
||||
{
|
||||
"id": "rv", "머리": ["돌기슭막이", "(형태마다)", "H=(높이마다)"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 기슭막이 · 관 유입·유출 기슭막이 길이 · 형태 → 높이마다 열",
|
||||
"일위대가": "UA000004",
|
||||
"바인딩": {
|
||||
"종류": "revetment", "펼침": ["form", "height_m"], "값": "length_m", "묶음": "rv",
|
||||
"머리틀": ["돌기슭막이", "{0}", "H={1}"],
|
||||
"이름": { "돌쌓기(찰)": "찰쌓기", "돌쌓기(메)": "메쌓기" },
|
||||
"더함": [
|
||||
{ "종류": "pipe", "펼침": ["inlet_revet_form", "inlet_revet_height_m"], "값": "inlet_revet_length_m", "조건": { "inlet_type": "기슭막이" } },
|
||||
{ "종류": "pipe", "펼침": ["outlet_revet_form", "outlet_revet_height_m"], "값": "outlet_revet_length_m", "조건": { "outlet_type": "기슭막이" } }
|
||||
]
|
||||
}
|
||||
},
|
||||
|
||||
{
|
||||
"id": "pv_b", "머리": ["콘크리트포장", "T=(두께마다)", "B"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 포장 폭",
|
||||
"바인딩": { "종류": "pavement_concrete", "펼침": ["thickness_cm"], "값": "width_m", "묶음": "pv", "머리틀": ["콘크리트포장", "T={0}cm", "B"] }
|
||||
},
|
||||
{
|
||||
"id": "pv_l", "머리": ["콘크리트포장", "T=(두께마다)", "L"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 포장 길이",
|
||||
"일위대가": "UA000007",
|
||||
"바인딩": { "종류": "pavement_concrete", "펼침": ["thickness_cm"], "값": "길이", "묶음": "pv", "머리틀": ["콘크리트포장", "T={0}cm", "L"] }
|
||||
},
|
||||
{
|
||||
"id": "pv_w", "머리": ["콘크리트포장", "T=(두께마다)", "확폭"], "단위": "㎡", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 확폭 면적",
|
||||
"바인딩": { "종류": "pavement_concrete", "펼침": ["thickness_cm"], "값": "widening_area_m2", "묶음": "pv", "머리틀": ["콘크리트포장", "T={0}cm", "확폭"] }
|
||||
},
|
||||
{
|
||||
"id": "pv_a", "머리": ["콘크리트포장", "T=(두께마다)", "A"], "단위": "㎡", "꼴": "수", "펼침": true,
|
||||
"식": "IF([pv_l]>0,[pv_b]*[pv_l]+[pv_w],\"\")",
|
||||
"끝수": { "자리": 2, "방법": "반올림" },
|
||||
"설명": "계산 — B × L + 확폭",
|
||||
"일위대가": "UA000006",
|
||||
"바인딩": { "종류": "pavement_concrete", "펼침": ["thickness_cm"], "값": "식", "묶음": "pv", "머리틀": ["콘크리트포장", "T={0}cm", "A"] }
|
||||
},
|
||||
{
|
||||
"id": "pv_jt", "머리": ["콘크리트포장", "T=(두께마다)", "수축줄눈"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"식": "IF([pv_l]>0,INT([pv_l]/[$수축줄눈_간격_m])*[pv_b],\"\")",
|
||||
"끝수": { "자리": 2, "방법": "반올림" },
|
||||
"설명": "계산 — 내림(L ÷ 줄눈 간격) × B · 줄눈 간격 = 설계값(기본 6m · 다르면 그 줄 식에 박음)",
|
||||
"바인딩": { "종류": "pavement_concrete", "펼침": ["thickness_cm"], "값": "식", "묶음": "pv", "머리틀": ["콘크리트포장", "T={0}cm", "수축줄눈"] }
|
||||
},
|
||||
|
||||
{
|
||||
"id": "pp_len", "머리": ["관공", "(관종·관경마다)", "관매설"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 횡단 관 연장(B06) · 같은 측점에 관이 둘이면 줄을 나눔",
|
||||
"일위대가": "UA000036",
|
||||
"일위대가후보": ["UA000036", "UA000027"],
|
||||
"바인딩": { "종류": "pipe", "펼침": ["pipe_kind", "pipe_diameter_mm"], "값": "관연장", "묶음": "pp", "머리틀": ["관공", "{0} Φ{1}", "관매설"] }
|
||||
},
|
||||
{
|
||||
"id": "pp_cp", "머리": ["관공", "(관종·관경마다)", "커플링밴드"], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"식": "IF([pp_len]>0,ROUNDUP([pp_len]/[$파형강관_1본_m],0)-1,\"\")",
|
||||
"설명": "계산 — 올림(L ÷ 1본 길이) − 1 · 관마다(줄마다) 계산 뒤 합 · 1본 길이 = 변수(관종마다 · 빈칸이면 계산 안 함)",
|
||||
"바인딩": {
|
||||
"종류": "pipe", "펼침": ["pipe_kind", "pipe_diameter_mm"], "값": "식", "묶음": "pp", "머리틀": ["관공", "{0} Φ{1}", "커플링밴드"],
|
||||
"식틀": "IF([pp_len]>0,ROUNDUP([pp_len]/[${0}_1본_m],0)-1,\"\")"
|
||||
}
|
||||
},
|
||||
|
||||
{
|
||||
"id": "pg_in", "머리": ["관보호공", "(형식마다)", "유입"], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 관 날개벽 형식 · 유입구가 집수정이면 집수정 형식 · 형식마다 열(유출구는 기슭막이 — 돌기슭막이 열)",
|
||||
"일위대가": "UA000008",
|
||||
"일위대가후보": ["UA000008", "UA000025", "UA000029"],
|
||||
"바인딩": {
|
||||
"종류": "pipe", "펼침": ["wing_wall_type"], "값": "개소", "묶음": "pg", "머리틀": ["관보호공", "{0}", "유입"],
|
||||
"이름": { "A-TYPE": "날개벽 A형", "C-TYPE": "날개벽 C형", "A-TYPE+집수정": "날개벽 A형+집수정" },
|
||||
"더함": [ { "종류": "pipe", "펼침": ["inlet_basin_form"], "값": "개소", "조건": { "inlet_type": "집수정" } } ]
|
||||
}
|
||||
},
|
||||
|
||||
{
|
||||
"id": "bx_len", "머리": ["BOX암거", "(폭×높이마다)", "연장"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — BOX 연장(계류 방향)",
|
||||
"일위대가": "UA000014",
|
||||
"바인딩": { "종류": "box_culvert", "펼침": ["body_width_m", "body_height_m"], "값": "length_m", "묶음": "bx", "머리틀": ["BOX암거", "{0}×{1}", "연장"] }
|
||||
},
|
||||
{
|
||||
"id": "bx_wing", "머리": ["BOX암거", "(폭×높이마다)", "날개벽"], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 날개벽 유입·유출 있음 수",
|
||||
"일위대가": "UA000025",
|
||||
"바인딩": {
|
||||
"종류": "box_culvert", "펼침": ["body_width_m", "body_height_m"], "값": "개소", "묶음": "bx", "머리틀": ["BOX암거", "{0}×{1}", "날개벽"],
|
||||
"조건": { "wing_in": "있음" },
|
||||
"더함": [ { "종류": "box_culvert", "펼침": ["body_width_m", "body_height_m"], "값": "개소", "조건": { "wing_out": "있음" } } ]
|
||||
}
|
||||
},
|
||||
|
||||
{
|
||||
"id": "fb", "머리": ["세월교", "(관경 × 련수마다)", null], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 세월교 개소",
|
||||
"바인딩": { "종류": "ford_bridge", "펼침": ["pipe_diameter_mm", "pipe_count"], "값": "개소", "묶음": "fb", "머리틀": ["세월교", "Φ{0}×{1}련", null] }
|
||||
},
|
||||
|
||||
{
|
||||
"id": "fp_w", "머리": ["물넘이포장", "T=(두께마다)", "폭"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 월류 폭",
|
||||
"바인딩": { "종류": "ford_pavement", "펼침": ["thickness_cm"], "값": "ford_width_m", "묶음": "fp", "머리틀": ["물넘이포장", "T={0}cm", "폭"] }
|
||||
},
|
||||
{
|
||||
"id": "fp_l", "머리": ["물넘이포장", "T=(두께마다)", "길이"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 포장 길이(노폭 방향)",
|
||||
"바인딩": { "종류": "ford_pavement", "펼침": ["thickness_cm"], "값": "length_m", "묶음": "fp", "머리틀": ["물넘이포장", "T={0}cm", "길이"] }
|
||||
},
|
||||
|
||||
{
|
||||
"id": "od_len", "머리": ["횡단개거", "(규격마다)", "길이"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 개거 연장",
|
||||
"일위대가": "UA000026",
|
||||
"바인딩": { "종류": "open_ditch", "펼침": ["ditch_spec"], "값": "length_m", "묶음": "od", "머리틀": ["횡단개거", "{0}", "길이"] }
|
||||
},
|
||||
{
|
||||
"id": "od_cnt", "머리": ["횡단개거", "(규격마다)", "개소"], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 개거 · 노출형 횡단수로 개소",
|
||||
"일위대가": "UA000003",
|
||||
"바인딩": {
|
||||
"종류": "open_ditch", "펼침": ["ditch_spec"], "값": "개소", "묶음": "od", "머리틀": ["횡단개거", "{0}", "개소"],
|
||||
"더함": [ { "종류": "cross_drain_exposed", "펼침": ["=노출형"], "값": "개소" } ]
|
||||
}
|
||||
},
|
||||
|
||||
{
|
||||
"id": "rw", "머리": ["옹벽", "(형식마다)", "H=(높이마다)"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 옹벽 길이",
|
||||
"바인딩": { "종류": "retaining_wall", "펼침": ["form", "height_m"], "값": "길이", "묶음": "rw", "머리틀": ["옹벽", "{0}", "H={1}"] }
|
||||
},
|
||||
{
|
||||
"id": "ms", "머리": ["돌쌓기", "(찰·메마다)", "H=(높이마다)"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 돌쌓기 길이",
|
||||
"바인딩": {
|
||||
"종류": "masonry_wet", "펼침": ["=찰쌓기", "height_m"], "값": "길이", "묶음": "ms", "머리틀": ["돌쌓기", "{0}", "H={1}"],
|
||||
"더함": [ { "종류": "masonry_dry", "펼침": ["=메쌓기", "height_m"], "값": "길이" } ]
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "sg", "머리": ["흙막이", "(형태마다)", "H=(높이마다)"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 흙막이 길이 · 일위대가는 큰돌만",
|
||||
"일위대가": "UA000032",
|
||||
"바인딩": { "종류": "soil_guard", "펼침": ["form", "height_m"], "값": "길이", "묶음": "sg", "머리틀": ["흙막이", "{0}", "H={1}"] }
|
||||
},
|
||||
{
|
||||
"id": "bm_len", "머리": ["큰돌쌓기", "(찰·메 · 돌 크기 · 높이마다)", "길이"], "단위": "m", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 큰돌쌓기 길이",
|
||||
"바인딩": { "종류": "boulder_masonry", "펼침": ["bond", "stone_cm", "height_m"], "값": "길이", "묶음": "bm", "머리틀": ["큰돌쌓기", "{0} {1} H={2}", "길이"] }
|
||||
},
|
||||
{
|
||||
"id": "bm_area", "머리": ["큰돌쌓기", "(찰·메 · 돌 크기 · 높이마다)", "면적"], "단위": "㎡", "꼴": "수", "펼침": true,
|
||||
"설명": "계산 — L × H (H = 펼친 높이)",
|
||||
"일위대가": "UA000031",
|
||||
"바인딩": {
|
||||
"종류": "boulder_masonry", "펼침": ["bond", "stone_cm", "height_m"], "값": "식", "묶음": "bm", "머리틀": ["큰돌쌓기", "{0} {1} H={2}", "면적"],
|
||||
"식틀": "IF([bm_len]>0,[bm_len]*{2},\"\")"
|
||||
}
|
||||
},
|
||||
|
||||
{
|
||||
"id": "ec", "머리": ["골막이", "(형태마다)", null], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 골막이 개소",
|
||||
"바인딩": { "종류": "erosion_check", "펼침": ["form"], "값": "개소", "묶음": "ec", "머리틀": ["골막이", "{0}", null] }
|
||||
},
|
||||
{
|
||||
"id": "be", "머리": ["소단", null, null], "단위": "m", "꼴": "수",
|
||||
"설명": "설계값 — 소단 길이",
|
||||
"바인딩": { "종류": "berm", "값": "길이" }
|
||||
},
|
||||
{ "id": "rf", "머리": ["대피소", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 대피소", "바인딩": { "종류": "refuge", "값": "개소" } },
|
||||
{ "id": "wy", "머리": ["정차·작업장", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 정차·작업장", "바인딩": { "종류": "work_yard", "값": "개소" } },
|
||||
{ "id": "ta", "머리": ["차돌림곳", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 차돌림곳", "바인딩": { "종류": "turnaround", "값": "개소" } },
|
||||
{ "id": "gr_rail", "머리": ["가드레일", null, null], "단위": "m", "꼴": "수", "설명": "설계값 — 가드레일 길이", "바인딩": { "종류": "guardrail", "값": "길이", "조건": { "kind": "가드레일" } } },
|
||||
{ "id": "gr_curb", "머리": ["경계석", null, null], "단위": "m", "꼴": "수", "설명": "설계값 — 경계석 길이", "바인딩": { "종류": "guardrail", "값": "길이", "조건": { "kind": "경계석" } } },
|
||||
{ "id": "gr_sign", "머리": ["위험표지", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 위험표지", "일위대가": "UA000017", "바인딩": { "종류": "guardrail", "값": "개소", "조건": { "kind": "위험표지" } } },
|
||||
{ "id": "mr", "머리": ["반사경", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 반사경", "일위대가": "UA000018", "바인딩": { "종류": "mirror", "값": "개소" } },
|
||||
{ "id": "bg", "머리": ["차단기", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 차단기", "일위대가": "UA000010", "바인딩": { "종류": "barrier_gate", "값": "개소" } },
|
||||
{ "id": "ps", "머리": ["국가지점번호판", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 번호판 · 전구간 줄은 손 입력", "일위대가": "UA000016", "바인딩": { "종류": "position_sign", "값": "개소" } },
|
||||
{ "id": "cs", "머리": ["준공표지석", null, null], "단위": "개소", "꼴": "수", "설명": "설계값 — 준공표지판", "일위대가": "UA000002", "바인딩": { "종류": "completion_sign", "값": "개소" } },
|
||||
{
|
||||
"id": "etc", "머리": ["기타", "(이름마다)", null], "단위": "개소", "꼴": "수", "펼침": true,
|
||||
"설명": "설계값 — 기타 구조물 개소",
|
||||
"바인딩": { "종류": "etc", "펼침": ["name"], "값": "개소", "묶음": "etc", "머리틀": ["기타", "{0}", null] }
|
||||
},
|
||||
|
||||
{ "id": "h_intake", "머리": ["취수정", null, null], "단위": "개소", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000005" },
|
||||
{ "id": "h_rip_c", "머리": ["돌붙임", "찰붙임", null], "단위": "㎡", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000009" },
|
||||
{ "id": "h_rip_m", "머리": ["돌붙임", "메붙임", null], "단위": "㎡", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000009" },
|
||||
{ "id": "h_stone_ch", "머리": ["돌수로", null, null], "단위": "m", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000011" },
|
||||
{ "id": "h_stone_bed", "머리": ["돌조공", null, null], "단위": "m", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000023" },
|
||||
{ "id": "h_pv_rail", "머리": ["포장 난간", null, null], "단위": "m", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000012" },
|
||||
{ "id": "h_pg_rail", "머리": ["관보호공 안전난간", null, null], "단위": "개소", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000013" },
|
||||
{ "id": "h_soil_ditch", "머리": ["수로형 토사개거", null, null], "단위": "개소", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000033" },
|
||||
{ "id": "h_ford_basin", "머리": ["물넘이 집수부", null, null], "단위": "개소", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": "UA000034" },
|
||||
{ "id": "h_rockfall", "머리": ["낙석방지책", null, null], "단위": "경간", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": null },
|
||||
{ "id": "h_break", "머리": ["콘크리트·포장 깨기", null, null], "단위": "㎥", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": null },
|
||||
{ "id": "h_waste", "머리": ["폐기물처리", null, null], "단위": "ton", "꼴": "수", "손": true, "설명": "손 입력", "일위대가": null },
|
||||
|
||||
{ "id": "memo", "머리": ["비고", null, null], "단위": null, "꼴": "글", "손": true, "설명": "손 입력" }
|
||||
],
|
||||
"줄": [
|
||||
{ "id": "all", "값": { "sta": "전구간" }, "고정": "전구간" }
|
||||
],
|
||||
"합계줄": [
|
||||
{ "id": "sum", "이름": "합계", "식": "SUM" }
|
||||
],
|
||||
"보기": { "틀고정": { "열": 2 }, "열너비": { "no": 40, "sta": 150, "memo": 120 } }
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
"""M02 양식 층 — 자리 · 복사 · 초기화 · 회사 적용 · 가져오기 · 권한 (임시 storage)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from common_util import common_util_storage
|
||||
from common_util.common_util_auth import verify_session
|
||||
from config import config_system
|
||||
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as router_module
|
||||
from M02_MasterTemplete import M02_Template_Layers as layers
|
||||
|
||||
P1 = "11111111-1111-1111-1111-111111111111"
|
||||
P2 = "22222222-2222-2222-2222-222222222222"
|
||||
P_OTHER = "33333333-3333-3333-3333-333333333333"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
||||
storage = tmp_path / "storage"
|
||||
storage.mkdir()
|
||||
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
|
||||
monkeypatch.setattr(common_util_storage, "STORAGE_BASE_DIR", str(storage))
|
||||
monkeypatch.setattr(layers, "SYSTEM_ROOT", tmp_path / "master_template")
|
||||
layers.write_template(layers.system_dir(), "table", "구조물집계표", {"판": 1, "열": []})
|
||||
layers.write_template(layers.system_dir(), "drawing", "A1_도각", {"format": 6})
|
||||
|
||||
projects = {
|
||||
P1: {"id": P1, "name": "첫째", "company_id": 7, "user_id": 42},
|
||||
P2: {"id": P2, "name": "둘째", "company_id": 7, "user_id": 43},
|
||||
P_OTHER: {"id": P_OTHER, "name": "남의 회사", "company_id": 9, "user_id": 90},
|
||||
}
|
||||
for row in projects.values():
|
||||
row["storage_path"] = f"storage/{row['company_id']}/{row['user_id']}/{row['id']}"
|
||||
root = Path(common_util_storage.resolve_stored_project_path(row["storage_path"]))
|
||||
layers.seed_project(root)
|
||||
users = {42: 7, 43: 7, 90: 9}
|
||||
|
||||
async def project_row(project_id: str) -> dict[str, Any] | None:
|
||||
found = projects.get(str(project_id))
|
||||
return dict(found) if found else None
|
||||
|
||||
async def user_company(user_id: int) -> int | None:
|
||||
return users.get(user_id)
|
||||
|
||||
async def company_users(company_id: int) -> list[dict[str, Any]]:
|
||||
return [{"user_id": u, "name": f"u{u}"} for u, c in users.items() if c == company_id]
|
||||
|
||||
async def company_projects(company_id: int) -> list[dict[str, Any]]:
|
||||
return [
|
||||
{"project_id": p["id"], "name": p["name"], "storage_path": p["storage_path"]}
|
||||
for p in projects.values()
|
||||
if p["company_id"] == company_id
|
||||
]
|
||||
|
||||
monkeypatch.setattr(router_module, "_project_row", project_row)
|
||||
monkeypatch.setattr(router_module, "_user_company", user_company)
|
||||
monkeypatch.setattr(router_module, "_company_users", company_users)
|
||||
monkeypatch.setattr(router_module, "_company_projects", company_projects)
|
||||
|
||||
session = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
|
||||
app = FastAPI()
|
||||
app.include_router(router_module.router)
|
||||
app.dependency_overrides[verify_session] = lambda: session
|
||||
return {"client": TestClient(app), "session": session, "storage": storage}
|
||||
|
||||
|
||||
def _root(world: dict[str, Any], project_id: str, company: int = 7, user: int = 42) -> Path:
|
||||
return (world["storage"] / str(company) / str(user) / project_id).resolve()
|
||||
|
||||
|
||||
def _url(project_id: str = P1) -> str:
|
||||
return f"/api/m02/layers/project/templates/table/구조물집계표?project_id={project_id}"
|
||||
|
||||
|
||||
def test_프로젝트_만들기_복사는_작업본과_초기본_둘에_manifest(world: dict[str, Any]) -> None:
|
||||
root = _root(world, P1)
|
||||
assert (root / "templates/table/구조물집계표.json").is_file()
|
||||
assert (root / "templates/_initial/drawing/A1_도각.json").is_file()
|
||||
entry = layers.read_manifest(root / "templates")["table/구조물집계표"]
|
||||
assert entry["층"] == "system" and entry["판"] == layers.version_of(
|
||||
layers.system_dir() / "table/구조물집계표.json"
|
||||
)
|
||||
|
||||
|
||||
def test_옛_프로젝트_넣기는_더하기만(world: dict[str, Any]) -> None:
|
||||
root = _root(world, P1)
|
||||
layers.write_template(root / "templates", "table", "구조물집계표", {"고침": True})
|
||||
assert layers.seed_project(root, only_missing=True) == {"작업본": [], "초기": []}
|
||||
assert layers.read_template(root / "templates", "table", "구조물집계표")["문서"] == {
|
||||
"고침": True
|
||||
}
|
||||
|
||||
|
||||
def test_작업본_저장_판_다르면_409_초기화는_초기본으로(world: dict[str, Any]) -> None:
|
||||
client = world["client"]
|
||||
got = client.get(_url()).json()
|
||||
assert got["층"] == "project" and got["문서"]["판"] == 1
|
||||
saved = client.put(_url(), json={"판": got["판"], "문서": {"판": 2}})
|
||||
assert saved.status_code == 200
|
||||
stale = client.put(_url(), json={"판": got["판"], "문서": {"판": 3}})
|
||||
assert stale.status_code == 409
|
||||
reset = client.post(f"/api/m02/projects/{P1}/templates/reset", json={})
|
||||
assert "table/구조물집계표" in reset.json()["초기화"]
|
||||
assert client.get(_url()).json()["문서"] == {"판": 1, "열": []}
|
||||
initial = _root(world, P1) / "templates/_initial/table/구조물집계표.json"
|
||||
assert layers.read_template(initial.parent.parent, "table", "구조물집계표")["문서"]["판"] == 1
|
||||
|
||||
|
||||
def test_회사_공식_저장은_관리자만_적용은_작업본만(world: dict[str, Any]) -> None:
|
||||
client, session = world["client"], world["session"]
|
||||
client.put(_url(), json={"판": client.get(_url()).json()["판"], "문서": {"회사": 1}})
|
||||
body = {"to": "company", "종류": "table", "이름": "구조물집계표"}
|
||||
assert client.post(f"/api/m02/projects/{P1}/templates/save-as", json=body).status_code == 403
|
||||
session["role"] = "ADMIN"
|
||||
assert client.post(f"/api/m02/projects/{P1}/templates/save-as", json=body).status_code == 200
|
||||
applied = client.post(f"/api/m02/projects/{P2}/templates/apply", json={"from": "company"})
|
||||
assert applied.status_code == 200
|
||||
root2 = _root(world, P2, user=43)
|
||||
assert layers.read_template(root2 / "templates", "table", "구조물집계표")["문서"] == {"회사": 1}
|
||||
assert (
|
||||
layers.read_template(root2 / "templates/_initial", "table", "구조물집계표")["문서"]["판"]
|
||||
== 1
|
||||
)
|
||||
assert layers.read_manifest(root2 / "templates")["table/구조물집계표"]["층"] == "company"
|
||||
|
||||
|
||||
def test_개인_양식은_본인만_쓰고_같은_회사는_읽어_가져옴(world: dict[str, Any]) -> None:
|
||||
client, session = world["client"], world["session"]
|
||||
body = {"to": "personal", "종류": "table", "이름": "구조물집계표"}
|
||||
assert client.post(f"/api/m02/projects/{P1}/templates/save-as", json=body).status_code == 200
|
||||
mine = f"/api/m02/layers/personal/templates/table/구조물집계표?project_id={P1}"
|
||||
assert client.get(mine).status_code == 200
|
||||
session.update(user_id=43)
|
||||
theirs = mine + "&user_id=42"
|
||||
assert client.get(theirs).status_code == 200
|
||||
put = client.put(
|
||||
f"/api/m02/layers/personal/templates/table/구조물집계표?project_id={P1}&user_id=42",
|
||||
json={"판": None, "문서": {}},
|
||||
)
|
||||
assert put.status_code == 200 # user_id 는 쓰기에서 무시 — 본인(43) 자리에 새로 씀
|
||||
assert (world["storage"] / "7/43/templates/table/구조물집계표.json").is_file()
|
||||
got = client.post(
|
||||
f"/api/m02/projects/{P2}/templates/apply", json={"from": "personal", "user_id": 42}
|
||||
)
|
||||
assert got.status_code == 200
|
||||
assert client.get(mine.replace(P1, P_OTHER)).status_code == 403
|
||||
far = client.post(
|
||||
f"/api/m02/projects/{P2}/templates/apply", json={"from": "personal", "user_id": 90}
|
||||
)
|
||||
assert far.status_code == 403
|
||||
|
||||
|
||||
def test_가져오기는_같은_회사_프로젝트만_목록도(world: dict[str, Any]) -> None:
|
||||
client = world["client"]
|
||||
ok = client.post(
|
||||
f"/api/m02/projects/{P1}/templates/apply", json={"from": "project", "project_id": P2}
|
||||
)
|
||||
assert ok.status_code == 200
|
||||
bad = client.post(
|
||||
f"/api/m02/projects/{P1}/templates/apply", json={"from": "project", "project_id": P_OTHER}
|
||||
)
|
||||
assert bad.status_code == 403
|
||||
sources = client.get(f"/api/m02/projects/{P1}/sources").json()
|
||||
assert [row["project_id"] for row in sources["project"]] == [P2]
|
||||
assert {row["이름"] for row in sources["system"]} == {"구조물집계표", "A1_도각"}
|
||||
|
||||
|
||||
def test_시스템_층은_읽기만_이름은_막음(world: dict[str, Any]) -> None:
|
||||
client = world["client"]
|
||||
assert client.get("/api/m02/layers/system/templates").json()["양식"]
|
||||
put = client.put(
|
||||
"/api/m02/layers/system/templates/table/구조물집계표", json={"판": None, "문서": {}}
|
||||
)
|
||||
assert put.status_code == 403
|
||||
for name in ("..", "_initial", "a.b/c"):
|
||||
with pytest.raises(ValueError):
|
||||
layers.template_path(layers.system_dir(), "table", name)
|
||||
Reference in New Issue
Block a user