feat(회사에 올리기): 요청 장부 · 요청 · 목록 · 승인 · 거절 · 알림 봄 길 + 임시 storage 시험 (PLAN 706-1)

- 회사 층 폴더 올리기요청.json 하나 · DB 안 바꿈
- 승인 = 「회사로 복사…」와 같은 길(구조물 = 붙여넣기 · 사전조합 = 서고 복사 + 폴더 자리) · 원본 개인 것 그대로
- 권한 = M02_Layer_Access 재사용 · 일반 승인 403 · 다른 회사 장부 안 봄 · 같은 것 다시 요청 409

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

M02_MasterTemplete/M02_Share_Request.py

@@ -0,0 +1,243 @@
+"""회사에 올리기 요청 · 승인(PLAN 706-1 · 요구서 4장 6) — 일반 사용자가 개인 층 항목 ·
+사전조합을 회사 층에 올려 달라고 요청 · 회사 관리자가 승인(= 「회사로 복사…」와 같은 길 ·
+원본은 그대로) · 거절.
+
+장부 = 회사 층 폴더 `올리기요청.json` 하나(DB 안 씀) · `{줄: [요청 …]}`.
+요청 = `{id, 종류(item · precombo), 키, 이름, 경로(항목 폴더), 주인 C/U, 요청자, 요청자이름,
+글, 날짜, 상태(요청 · 승인 · 거절), 처리자, 처리날짜, 사유, 사본, 봄}` · `봄` = 요청자가 결과
+알림을 닫음. 권한 = `M02_Layer_Access` 한 벌 — 요청 = 자기 개인 층 · 목록 = 회사 관리자는
+그 회사 전부 · 일반은 자기 것 ·
+승인 · 거절 = 회사 층 쓰기(일반 403) · 다른 회사 장부는 안 봄(시스템 관리자만 `company`).
+같은 항목(주인 · 종류 · 키)이 「요청」 상태면 다시 요청 = 409 「요청 중」.
+⚠ 등록은 `main.py`(로그인만).
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Literal
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from common_util.common_util_auth import verify_session
+from common_util.common_util_json import atomic_write_json
+from M02_MasterTemplete import M02_Item_Paste as paste_store
+from M02_MasterTemplete import M02_Labor_Ref as refs
+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_Precombo_Folders as pre_folders
+from M02_MasterTemplete.M02_MasterTemplete_Router_Items import (
+    NotedRoute,
+    Session,
+    _call,
+    _root,
+)
+
+router = APIRouter(prefix="/api/m02", tags=["M02 Share Request"], route_class=NotedRoute)
+
+FILE = "올리기요청.json"
+# ponytail: 한 프로세스 잠금 — 서버를 여럿 띄우면 파일 잠금으로
+_LOCK = threading.Lock()
+
+
+class AskBody(BaseModel):
+    종류: Literal["item", "precombo"]
+    키: str
+    글: str = ""
+
+
+class ApproveBody(BaseModel):
+    경로: list[str] = []
+
+
+class RejectBody(BaseModel):
+    사유: str = ""
+
+
+def _company(session: Session, company: str | None) -> str:
+    """장부 회사 — 시스템 관리자만 `company` 로 남의 회사 · 그 밖은 자기 회사(없으면 409)."""
+    if company and access.is_system_admin(session):
+        return str(company)
+    own = session.get("company_id")
+    if own is None:
+        raise HTTPException(status_code=409, detail=access.NO_COMPANY)
+    return str(own)
+
+
+def _path(company: str) -> Path:
+    return access.folder(access.Grant("company", company, False, True)) / FILE
+
+
+def _read(company: str) -> list[dict[str, Any]]:
+    path = _path(company)
+    if not path.is_file():
+        return []
+    return list(json.loads(path.read_bytes()).get("줄") or [])
+
+
+def _write(company: str, rows: list[dict[str, Any]]) -> None:
+    atomic_write_json(_path(company), {"줄": rows})
+
+
+def _now() -> str:
+    return datetime.now().isoformat(timespec="seconds")
+
+
+def _source(session: Session, kind: str, key: str) -> dict[str, Any]:
+    """자기 개인 층 원본 — `{이름, 경로}`(없으면 404)."""
+    folder = _root(session, "personal")
+    if kind == "item":
+        found = next((r for r in items_store.items(folder) if r.get("key") == key), None)
+        if found is None:
+            raise HTTPException(status_code=404, detail=f"없는 항목 「{key}」")
+        path = list(found.get("경로") or [])
+        return {"이름": path[-1] if path else key, "경로": path[:-1]}
+    owner = access.own_owner(session, "personal")
+    doc = (
+        refs.read_ref({"층": "personal", "주인": owner, "키": key})
+        if refs._KEY.match(key)
+        else None
+    )
+    if doc is None or refs.kind_of(key) != "precombo":
+        raise HTTPException(status_code=404, detail=f"없는 사전조합 「{key}」")
+    return {"이름": str(doc.get("이름") or key), "경로": []}
+
+
+@router.post("/share-requests")
+def ask(body: AskBody, session: Session = Depends(verify_session)) -> dict[str, Any]:
+    """요청 — 자기 개인 층 것만 · 같은 것이 「요청」 이면 409 「요청 중」."""
+    owner = access.own_owner(session, "personal")
+    if owner is None:
+        raise HTTPException(status_code=409, detail=access.NO_COMPANY)
+    src = _source(session, body.종류, body.키)
+    company = owner.split("/")[0]
+    row = {
+        "id": uuid.uuid4().hex[:12],
+        "종류": body.종류,
+        "키": body.키,
+        **src,
+        "주인": owner,
+        "요청자": session.get("user_id"),
+        "요청자이름": session.get("name") or "",
+        "글": body.글.strip()[:200],
+        "날짜": _now(),
+        "상태": "요청",
+    }
+    with _LOCK:
+        rows = _read(company)
+        same = (owner, body.종류, body.키)
+        if any((r["주인"], r["종류"], r["키"]) == same and r["상태"] == "요청" for r in rows):
+            raise HTTPException(status_code=409, detail="요청 중 — 회사 관리자가 아직 안 봄")
+        _write(company, [*rows, row])
+    return row
+
+
+@router.get("/share-requests")
+def listing(
+    company: str | None = None, session: Session = Depends(verify_session)
+) -> dict[str, Any]:
+    """목록 — 회사를 맡은 사람 = 그 회사 전부 · 그 밖 = 자기 요청만 · `관리` = 승인할 수 있나."""
+    at = _company(session, company)
+    rows = _read(at)
+    boss = access.manages(session, at)
+    if not boss:
+        me = session.get("user_id")
+        rows = [r for r in rows if r.get("요청자") == me]
+    return {"회사": at, "관리": boss, "줄": rows}
+
+
+def _settle(session: Session, company: str | None, rid: str, done: dict[str, Any]) -> dict:
+    """요청 한 줄을 끝냄 — 회사 층 쓰기 권한(일반 403) · 「요청」 상태만(아니면 409)."""
+    at = _company(session, company)
+    access.require(session, "company", at, write=True)
+    with _LOCK:
+        rows = _read(at)
+        row = next((r for r in rows if r.get("id") == rid), None)
+        if row is None:
+            raise HTTPException(status_code=404, detail="없는 요청")
+        if row["상태"] != "요청":
+            raise HTTPException(status_code=409, detail=f"이미 {row['상태']}")
+        extra = done.pop("run", None)
+        if extra:
+            done["사본"] = extra(row, at)
+        row.update({**done, "처리자": session.get("name") or "", "처리날짜": _now(), "봄": False})
+        _write(at, rows)
+    return row
+
+
+def _copy(session: Session, dest: list[str]):
+    """승인 = 「회사로 복사…」와 같은 길 — 구조물 항목 = 붙여넣기 ·
+    사전조합 = 서고 복사(+ 폴더 자리)."""
+
+    def run(row: dict[str, Any], company: str) -> str:
+        target = _root(session, "company", company, write=True)
+        source = _root(session, "personal", row["주인"])
+        if row["종류"] == "item":
+            ref = [{"경로": row.get("경로") or [], "키": row["키"]}]
+            origin = {"layer": "personal", "owner": row["주인"]}
+            try:
+                note = paste_store.paste(source, target, ref, dest, origin)
+            except store.StoreError as e:
+                raise HTTPException(status_code=e.status, detail=e.detail) from e
+            return str(note["열"][0])
+        ref = {"층": "personal", "주인": row["주인"], "키": row["키"]}
+        doc = refs.read_ref(ref)
+        if doc is None:
+            raise HTTPException(status_code=404, detail=f"원본이 없음 「{row['키']}」")
+
+        def make() -> dict[
+            str, Any
+        ]:  # 못 읽는 참조 같이 복사 = 받는 층 안에서(`new_library` 와 같음)
+            return refs.create("precombo", refs.carry(refs.origin_mark(doc, ref), here))
+
+        here = ("personal", row["주인"])
+        made = _call(target, make)
+        if dest:
+            view = _call(target, pre_folders.view)
+            move = {"op": "move", "키": made["이름"], "새경로": dest, "판": view["판"]}
+            _call(target, pre_folders.edit, move)
+        return str(made["이름"])
+
+    return run
+
+
+@router.post("/share-requests/{rid}/approve")
+def approve(
+    rid: str,
+    body: ApproveBody,
+    company: str | None = None,
+    session: Session = Depends(verify_session),
+) -> dict[str, Any]:
+    """승인 — 회사 층 `경로` 폴더에 사본(새 키 · 같은 이름 「(1)」) · 원본 개인 것은 그대로."""
+    return _settle(session, company, rid, {"상태": "승인", "run": _copy(session, body.경로)})
+
+
+@router.post("/share-requests/{rid}/reject")
+def reject(
+    rid: str,
+    body: RejectBody,
+    company: str | None = None,
+    session: Session = Depends(verify_session),
+) -> dict[str, Any]:
+    return _settle(session, company, rid, {"상태": "거절", "사유": body.사유.strip()[:200]})
+
+
+@router.post("/share-requests/seen")
+def seen(session: Session = Depends(verify_session)) -> dict[str, int]:
+    """요청자가 결과 알림 줄을 닫음 — 자기 요청 가운데 끝난 것 `봄` = true."""
+    at = _company(session, None)
+    me = session.get("user_id")
+    with _LOCK:
+        rows = _read(at)
+        hit = [r for r in rows if r.get("요청자") == me and r.get("봄") is False]
+        for r in hit:
+            r["봄"] = True
+        if hit:
+            _write(at, rows)
+    return {"봄": len(hit)}

main.py

@@ -547,6 +547,11 @@ app.include_router(m02_items_router, dependencies=protected)
 app.include_router(m02_combos_router, dependencies=protected)
 app.include_router(m02_drawing_router, dependencies=protected)
 app.include_router(m02_labor_router, dependencies=protected)
+# 회사에 올리기 요청 · 승인(PLAN 706-1) — 로그인만 · 층 권한은 길 안에서
+app.include_router(
+    __import__("M02_MasterTemplete.M02_Share_Request", fromlist=["router"]).router,
+    dependencies=protected,
+)
 # 프로젝트 집계 항목 복사본(PLAN 25-2b) — 로그인만 · 프로젝트 권한은 길 안에서.
 app.include_router(m02_project_items_router, dependencies=protected)
 # 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근).

resources/tester/test_706_1_share_request.py

@@ -0,0 +1,129 @@
+"""회사에 올리기 요청 · 승인(PLAN 706-1 · `M02_Share_Request`) — 임시 storage.
+
+① 일반 사용자 요청(구조물 항목 · 사전조합) → 장부 한 줄 · 같은 것 다시 = 409 「요청 중」.
+② 회사 관리자 승인 = 회사 층 고른 폴더에 사본(원본 개인 것 그대로) · 거절 = 사유 · 요청자 알림 → 봄.
+③ 권한 — 일반 승인 403 · 다른 회사 관리자는 그 요청을 못 봄(404) · 남의 회사 장부 안 봄 ·
+   목록은 자기 것만.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from common_util.common_util_auth import verify_session
+from M02_MasterTemplete import M02_Labor_Router as labor_router
+from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router
+from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layers_router
+from M02_MasterTemplete import M02_Share_Request as share
+from resources.tester import test_m02_tree as base
+from resources.tester.test_m02_tree import Tree
+
+world = base.world  # 같은 임시 storage 판
+
+URL = "/api/m02/share-requests"
+
+
+def _client(world: dict[str, Any]) -> TestClient:
+    app = FastAPI()
+    for one in (layers_router, items_router, labor_router, share):
+        app.include_router(one.router)
+    app.dependency_overrides[verify_session] = lambda: world["session"]
+    return TestClient(app)
+
+
+def _as(world: dict[str, Any], role: str = "USER", user: int = 42, company: int = 7) -> None:
+    world["session"].update(role=role, user_id=user, company_id=company, name=f"사람{user}")
+
+
+def _mine(world: dict[str, Any], client: TestClient) -> tuple[str, str]:
+    """일반 사용자 42 개인 층 — 항목 「배수/sub7임시」 · 사전조합 하나 → (항목 키, 사전조합 키)."""
+    _as(world)
+    t = Tree(client, "personal")
+    t.do(op="add-folder", 경로=[], 이름="배수")
+    t.do(op="add-item", 경로=["배수"], 이름="sub7임시", 단위="m")
+    doc = {
+        "문서": {"이름": "sub7임시 사전", "구분": "C 사면안정", "상세구분": "옹벽", "담은로직": []}
+    }
+    pre = client.post("/api/m02/layers/personal/library/precombo", json=doc).json()["이름"]
+    return t.item(["배수", "sub7임시"])["key"], pre
+
+
+def test_요청_승인_거절_알림(world):
+    client = _client(world)
+    item, pre = _mine(world, client)
+    got = client.post(URL, json={"종류": "item", "키": item, "글": " 회사 공용으로 "})
+    assert got.status_code == 200, got.text
+    one = got.json()
+    assert (one["이름"], one["경로"], one["주인"], one["글"]) == (
+        "sub7임시",
+        ["배수"],
+        "7/42",
+        "회사 공용으로",
+    )
+    assert client.post(URL, json={"종류": "item", "키": item}).status_code == 409  # 요청 중
+    two = client.post(URL, json={"종류": "precombo", "키": pre}).json()
+    assert client.post(URL, json={"종류": "item", "키": "SP42-999999"}).status_code == 404
+    assert client.post(URL, json={"종류": "precombo", "키": "SC-999999"}).status_code == 404
+    mine = client.get(URL).json()
+    assert mine["관리"] is False and [r["id"] for r in mine["줄"]] == [one["id"], two["id"]]
+
+    _as(world, "ADMIN", user=50)  # 회사 7 관리자
+    company = Tree(client, "company")
+    company.do(op="add-folder", 경로=[], 이름="공용")
+    boss = client.get(URL).json()
+    assert boss["관리"] is True and len(boss["줄"]) == 2
+    ok = client.post(f"{URL}/{one['id']}/approve", json={"경로": ["공용"]})
+    assert ok.status_code == 200, ok.text
+    assert ok.json()["상태"] == "승인" and ok.json()["봄"] is False
+    company = Tree(client, "company")
+    assert ["공용", "sub7임시"] in company.paths()  # 고른 폴더에 사본
+    again = client.post(f"{URL}/{one['id']}/approve", json={"경로": []})
+    assert again.status_code == 409  # 이미 승인
+    no = client.post(f"{URL}/{two['id']}/reject", json={"사유": "이름 다시"})
+    assert no.json()["상태"] == "거절" and no.json()["사유"] == "이름 다시"
+
+    _as(world)  # 요청자 — 원본 그대로 · 결과 둘 · 봄 뒤 0
+    assert ["배수", "sub7임시"] in Tree(client, "personal").paths()
+    rows = client.get(URL).json()["줄"]
+    assert [(r["상태"], r["봄"]) for r in rows] == [("승인", False), ("거절", False)]
+    assert client.post(f"{URL}/seen").json() == {"봄": 2}
+    assert client.post(f"{URL}/seen").json() == {"봄": 0}
+    # 거절된 것은 다시 요청할 수 있음
+    assert client.post(URL, json={"종류": "precombo", "키": pre}).status_code == 200
+
+
+def test_사전조합_승인_폴더_자리(world):
+    client = _client(world)
+    _, pre = _mine(world, client)
+    rid = client.post(URL, json={"종류": "precombo", "키": pre}).json()["id"]
+    _as(world, "ADMIN", user=50)
+    tree_url = "/api/m02/layers/company/precombo-tree"
+    now = client.get(tree_url).json()
+    client.post(tree_url, json={"op": "add-folder", "경로": [], "이름": "옹벽", "판": now["판"]})
+    made = client.post(f"{URL}/{rid}/approve", json={"경로": ["옹벽"]}).json()["사본"]
+    assert client.get(tree_url).json()["자리"] == {made: ["옹벽"]}
+    doc = client.get(f"/api/m02/layers/company/templates/precombo/{made}").json()["문서"]
+    assert doc["이름"] == "sub7임시 사전"
+    assert doc["본뜬"]["주인"] == "7/42"  # 원본 표시
+
+
+def test_권한_일반_승인_403_다른_회사(world):
+    client = _client(world)
+    item, _ = _mine(world, client)
+    rid = client.post(URL, json={"종류": "item", "키": item}).json()["id"]
+    assert client.post(f"{URL}/{rid}/approve", json={}).status_code == 403  # 일반 = 승인 못 함
+    assert client.post(f"{URL}/{rid}/reject", json={}).status_code == 403
+    _as(world, "USER", user=43)  # 같은 회사 다른 일반 — 남의 요청 안 보임
+    assert client.get(URL).json()["줄"] == []
+    _as(world, "ADMIN", user=90, company=9)  # 다른 회사 관리자
+    assert client.get(URL).json()["줄"] == []
+    assert client.post(f"{URL}/{rid}/approve", json={}).status_code == 404  # 자기 장부에 없음
+    assert client.get(URL, params={"company": "7"}).json()["회사"] == "9"  # 남의 회사 고르기 무시
+    _as(world, "SYSTEM_ADMIN", user=1, company=None)
+    assert len(client.get(URL, params={"company": "7"}).json()["줄"]) == 1
+    ledger = world["storage"] / "7" / "templates" / share.FILE
+    assert json.loads(ledger.read_bytes())["줄"][0]["상태"] == "요청"