feat(718-2): 별칭 장부 시스템 모으기 서버 — 회사 · 개인 층 맞춤 + 공사 내역맞춤 후보 · 익명 · 충돌 · 승인 → 마스터 · 반려 기억

- 새 M01_MasterData_Alias(/api/m01/alias-ledger · 시스템 관리자만 · main.py 한 줄) — 공유 storage 읽기만 · 같은 짝 묶음(서명) · 회사 수 · 사람 수만 · 같은 열쇠 다른 짝 · 마스터와 다름 = 충돌 · 승인 = 서버가 다시 모아 찾음 → 맞춤 ledger.put 재검사 · 내역맞춤 마스터 기억 · 반려 = 마스터 층 별칭반려 기억 · 자동 반영 없음
- B09 잇기 기억 읽기에 마스터 층을 맨 아래로(브레인 허락) · test_707_4 기억 층 시험에 마스터 줄 한 줄
- 새 시험 test_718_2_alias_gather(임시 storage — 모음 수 · 익명 · 충돌 · 승인 → 마스터 줄 · 다른 회사 새 사용자가 씀 · 반려 기억 · 권한)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-11-01 14:50:02 +09:00
co-authored by Claude Opus 5.5
parent 9827926a20
commit 28e9a2a667
5 changed files with 361 additions and 3 deletions
+3 -2
View File
@@ -32,6 +32,7 @@ from B09_Estimation import B09_Estimation_Store as store
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_Layer_Access as access
from M02_MasterTemplete import M02_MasterTemplete_Store as m02_store
router = APIRouter()
PATH = "/{project_id}/estimation/link"
@@ -51,8 +52,8 @@ def _folders(session: dict[str, Any]) -> dict[str, tuple[Path, bool]]:
def recalled(session: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""맞춤 기억 — 회사 위에 개인을 덮음."""
out: dict[str, dict[str, Any]] = {}
"""맞춤 기억 — 마스터(시스템 관리자 승인 · 718-2) 위에 회사 · 그 위에 개인을 덮음."""
out: dict[str, dict[str, Any]] = dict(hopyo_store.memory(m02_store.FOLDER, linker.MEMORY))
for folder, _write in _folders(session).values():
out.update(hopyo_store.memory(folder, linker.MEMORY))
return out
+215
View File
@@ -0,0 +1,215 @@
"""M01 별칭 장부 시스템 모으기(PLAN 718-2 · 요구서 내역12 4장 23 · 4-1 확정 6).
모든 회사 · 개인 층의 별칭 장부(`실무자료/맞춤.json` · `M02_Hopyo_Ledger`)와 B09 공사 내역 맞춤
기억(`실무자료/내역맞춤.json` · 707-4)을 읽어 후보로 모음 — 읽기만(공유 storage 안 씀).
후보 = (종류, 열쇠, 서명) · 서명 = 짝 내용 지문(같은 짝이면 같은 서명) · 회사 수 · 사람 수만
(누가 · 처음파일 · 층 주인은 안 내보냄 = 익명) · 충돌 = 같은 열쇠 다른 서명 · 마스터와 다름.
승인 = 시스템 관리자만 → 마스터 층에 씀(맞춤 = `ledger.put` 재검사 · 내역맞춤 = 기억 파일) ·
반려 = 마스터 층 `실무자료/별칭반려.json` 에 서명 기억(같은 짝은 다시 안 보임) · 자동 반영 없음.
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from common_util.common_util_auth import verify_session
from common_util.common_util_time import utc_now_text
from M02_MasterTemplete import M02_Hopyo_Ledger as ledger
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers
ALIAS, BILL = "맞춤", "내역맞춤"
KINDS = (ALIAS, BILL)
REJECT = "별칭반려"
router = APIRouter(prefix="/api/m01/alias-ledger", tags=["M01 별칭 장부"])
# ── 읽기 ─────────────────────────────────────────────────────────────
def layer_folders() -> list[tuple[str, str | None, Path]]:
"""(회사, 사용자 | None, 층 폴더) — storage 의 회사 · 개인 층(숫자 폴더만 · 두 단만 훑음)."""
root = layers.storage_root()
out: list[tuple[str, str | None, Path]] = []
for c in sorted(root.iterdir() if root.is_dir() else []):
if not (c.is_dir() and c.name.isdigit()):
continue
out.append((c.name, None, layers.company_dir(c.name)))
for u in sorted(c.iterdir()):
if u.is_dir() and u.name.isdigit():
out.append((c.name, u.name, layers.personal_dir(c.name, u.name)))
return out
def _body(kind: str, row: dict[str, Any]) -> dict[str, Any]:
"""서명에 드는 내용 — 맞춤 = 갈래 · 짝 · 내역맞춤 = 확정 줄 키 · 이름."""
if kind == ALIAS:
return {"갈래": row.get("갈래"), "짝": row.get("짝") or []}
return {"확정": row.get("확정"), "이름": row.get("이름")}
def sign(kind: str, row: dict[str, Any]) -> str:
raw = json.dumps(_body(kind, row), ensure_ascii=False, sort_keys=True)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
def _book(kind: str, folder: Path) -> dict[str, dict[str, Any]]:
if kind == ALIAS:
return ledger.read(folder)
got = hopyo_store.memory(folder, BILL)
return {
k: v
for k, v in got.items()
if isinstance(v, dict) and v.get("확정") and str(k).count("|") == 2
} # 잊음 · 깨진 열쇠 뺌
def _origin(full: str, row: dict[str, Any]) -> dict[str, str]:
if isinstance(row.get("원래"), dict):
return {k: str(row["원래"].get(k) or "") for k in ("품명", "규격", "단위")}
return ledger._plain_of(full)
def rejected() -> dict[str, dict[str, Any]]:
return hopyo_store.memory(store.FOLDER, REJECT)
def _reject_id(kind: str, full: str, sig: str) -> str:
return f"{kind}|{sig}|{full}"
def gather() -> dict[str, Any]:
"""후보 모음 `{후보, 모은곳{회사, 사람}, 반려}` — 마스터와 같은 짝 · 반려한 짝은 뺌."""
seen: dict[tuple[str, str, str], dict[str, Any]] = {}
firms, people = set(), set()
for company, user, folder in layer_folders():
for kind in KINDS:
for full, row in _book(kind, folder).items():
sig = sign(kind, row)
who = (
(row.get("누가") or {}).get("id") if isinstance(row.get("누가"), dict) else None
)
person = f"{company}/{user}" if user else f"{company}/{who or '-'}"
one = seen.setdefault(
(kind, full, sig),
{"종류": kind, "열쇠": full, "서명": sig, "원래": _origin(full, row),
**_body(kind, row), "회사": set(), "사람": set(), "때": None},
) # fmt: skip
one["회사"].add(company)
one["사람"].add(person)
firms.add(company)
people.add(person)
stamp = row.get("때")
if stamp and (one["때"] is None or str(stamp) > str(one["때"])):
one["때"] = stamp
master = {kind: _book(kind, Path(store.FOLDER)) for kind in KINDS}
gone = rejected()
live = []
for (kind, full, sig), one in seen.items():
mine = master[kind].get(full)
if (mine and sign(kind, mine) == sig) or _reject_id(kind, full, sig) in gone:
continue
one["마스터"] = "다름" if mine else None
live.append(one)
many: dict[tuple[str, str], int] = {}
for one in live:
many[(one["종류"], one["열쇠"])] = many.get((one["종류"], one["열쇠"]), 0) + 1
out = []
for one in sorted(live, key=lambda x: (x["종류"], x["열쇠"], -len(x["사람"]), x["서명"])):
clash = many[(one["종류"], one["열쇠"])] > 1 or one["마스터"] == "다름"
out.append({**one, "회사": len(one["회사"]), "사람": len(one["사람"]), "충돌": clash})
return {"후보": out, "모은곳": {"회사": len(firms), "사람": len(people)}, "반려": len(gone)}
def _find(kind: str, full: str, sig: str) -> dict[str, Any]:
if kind not in KINDS:
raise store.StoreError(400, f"종류는 {' · '.join(KINDS)}")
got = next(
(x for x in gather()["후보"] if (x["종류"], x["열쇠"], x["서명"]) == (kind, full, sig)),
None,
)
if got is None:
raise store.StoreError(404, f"후보가 없음(이미 승인 · 반려 · 바뀜) 「{full}」")
return got
# ── 쓰기(마스터 층만) ─────────────────────────────────────────────────
def approve(kind: str, full: str, sig: str, who: Any) -> dict[str, Any]:
"""후보 하나 → 마스터 장부(서버가 다시 모아 찾음 · 맞춤은 `ledger.put` 재검사)."""
one = _find(kind, full, sig)
if kind == ALIAS:
ledger.put(store.FOLDER, {full: {"원래": one["원래"], **_body(kind, one)}}, who)
else:
row = {**_body(kind, one), "누가": who, "때": utc_now_text()}
hopyo_store.remember(store.FOLDER, {full: row}, BILL)
return gather()
def reject(kind: str, full: str, sig: str, who: Any) -> dict[str, Any]:
_find(kind, full, sig)
row = {"종류": kind, "열쇠": full, "서명": sig, "누가": who, "때": utc_now_text()}
hopyo_store.remember(store.FOLDER, {_reject_id(kind, full, sig): row}, REJECT)
return gather()
def master_bill() -> list[dict[str, Any]]:
"""마스터 내역맞춤 줄(보기 · 지움) — 맞춤 장부는 `/api/m02/layers/system/hopyo-ledger`."""
book = _book(BILL, Path(store.FOLDER))
return [
{"열쇠": k, "원래": ledger._plain_of(k), **_body(BILL, v)} for k, v in sorted(book.items())
]
def forget_bill(full: str) -> list[dict[str, Any]]:
if full not in _book(BILL, Path(store.FOLDER)):
raise store.StoreError(404, f"마스터 내역맞춤에 없음 「{full}」")
hopyo_store.remember(store.FOLDER, {full: None}, BILL)
return master_bill()
# ── 길(시스템 관리자만 — main.py 에서 system_admin_only) ───────────────
def _who(session: dict[str, Any]) -> dict[str, Any]:
return {"id": session.get("user_id"), "이름": session.get("name")}
def _call(fn, *args):
try:
return fn(*args)
except store.StoreError as error:
raise HTTPException(error.status, error.detail) from error
class Pick(BaseModel):
종류: str = Field(max_length=10)
열쇠: str = Field(min_length=1, max_length=400)
서명: str = Field(min_length=16, max_length=16)
@router.get("")
def candidates() -> dict[str, Any]:
return {**gather(), "마스터내역": master_bill()}
@router.post("/approve")
def approve_one(body: Pick, session: dict = Depends(verify_session)) -> dict[str, Any]:
return _call(approve, body.종류, body.열쇠, body.서명, _who(session))
@router.post("/reject")
def reject_one(body: Pick, session: dict = Depends(verify_session)) -> dict[str, Any]:
return _call(reject, body.종류, body.열쇠, body.서명, _who(session))
class Forget(BaseModel):
열쇠: str = Field(min_length=1, max_length=400)
@router.post("/master-bill/forget")
def forget_one(body: Forget) -> dict[str, Any]:
return {"마스터내역": _call(forget_bill, body.열쇠)}
+4
View File
@@ -544,6 +544,10 @@ app.include_router(b09_estimation_router, dependencies=protected_with_company)
# 마스터 데이터 — 회사·프로젝트가 아니라 시스템 관리자만 본다.
system_admin_only = [Depends(verify_session), Depends(require_system_admin)]
app.include_router(m01_master_data_router, dependencies=system_admin_only)
app.include_router(
__import__("M01_MasterData.M01_MasterData_Alias", fromlist=["router"]).router,
dependencies=system_admin_only,
) # 별칭 장부 모으기(718-2)
# 단가산출 흐름도 시험 길 — 폴더째 지우면 이 줄도 지움
app.include_router(
__import__("M02_MasterTemplete.DataFlow.M02_DF_Router", fromlist=["router"]).router,
+5 -1
View File
@@ -211,6 +211,9 @@ def test_못_잇는_짝(geo):
)
def test_기억_층(tmp_path, monkeypatch, role, layer):
monkeypatch.setattr(route.access, "folder", lambda got, create=False: tmp_path / got.layer)
monkeypatch.setattr(route.m02_store, "FOLDER", tmp_path / "system")
master = {"a||": {"확정": "earth:9"}, "m||": {"확정": "earth:1"}} # 시스템 승인(718-2)
hopyo_store.remember(tmp_path / "system", master, lk.MEMORY)
session = {"company_id": 7, "user_id": 3, "role": role}
assert route.keep(session, {"a||": {"확정": "earth:4", "이름": "절토(토사)"}}) == layer
assert route.keep(session, {}) is None
@@ -220,7 +223,8 @@ def test_기억_층(tmp_path, monkeypatch, role, layer):
got = route.recalled(session)
assert got["a||"]["확정"] == ("earth:4" if layer == "personal" else "earth:5")
assert got["b||"]["확정"] == "x"
assert route.recalled({"user_id": 3, "role": role}) == {} # 회사 없음 = 층 없음
assert got["m||"]["확정"] == "earth:1" # 회사 · 개인 줄이 마스터보다 앞섬(a = 아래 층)
assert route.recalled({"user_id": 3, "role": role}) == master # 회사 없음 = 마스터만
def _same(a, b):
+134
View File
@@ -0,0 +1,134 @@
"""별칭 장부 시스템 모으기(PLAN 718-2 · `M01_MasterData_Alias`) — 회사 · 개인 층 맞춤 +
내역맞춤 모음 · 같은 짝 묶기 · 회사 수 · 사람 수(익명) · 충돌 · 승인 → 마스터 줄 ·
반려 기억 · 다른 회사 새 사용자가 마스터 줄을 씀
· 시스템 관리자만 · master_data 원본 그대로. 층 · storage 는 임시 폴더.
"""
import json
from pathlib import Path
import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from B09_Estimation import B09_Estimation_Bill_Link as lk
from B09_Estimation import B09_Estimation_Router_Link as route
from common_util.common_util_auth import require_system_admin, verify_session
from M01_MasterData import M01_MasterData_Alias as alias
from M02_MasterTemplete import M02_Hopyo_Ledger as ledger
from M02_MasterTemplete import M02_Hopyo_Store as hs
from M02_MasterTemplete import M02_Template_Layers as layers
# 올리기 길 시험의 `client` 마당(임시 storage · 마스터 층)을 같이 씀
# ruff: noqa: F811
from resources.tester.test_m02_hopyo_excel import client, master_hash # noqa: F401
DIG = {"이름": "구조물터파기(인력100%)", "규격": "토사", "단위": "㎥"}
DIG_KEY = "구조물터파기(인력100%)|토사|m3"
FORM_KEY = "합판거푸집|4회|m2"
CUT_KEY = "토사절취||m3"
ROOT = Path(__file__).resolve().parents[2]
def _pair(code):
return ledger.check(DIG_KEY, {"갈래": "짝", "짝": [{"로직": code, "배": 1}]})
def _seed():
"""회사 7(개인 42 · 43 = 같은 짝) · 회사 8(개인 50 = 다른 짝 · 회사 층 = 안 씀) · 내역맞춤."""
for user, who in ((42, "갑"), (43, "을")):
mine = layers.personal_dir(7, user)
ledger.write(mine, {DIG_KEY: _pair("GF000101")}, {"id": user, "이름": who}, "가공사.xls")
ledger.write(layers.personal_dir(8, 50), {DIG_KEY: _pair("GF000103")}, {"id": 50}, "나.xls")
skip = ledger.check(FORM_KEY, {"갈래": "안 씀"})
ledger.write(layers.company_dir(8), {FORM_KEY: skip}, {"id": 51, "이름": "병"}, "다.xls")
live = {"확정": "earth:4", "이름": "절토(토사)"}
hs.remember(layers.personal_dir(7, 42), {CUT_KEY: live, "a||": {lk.FORGOT: True}}, lk.MEMORY)
hs.remember(layers.company_dir(8), {CUT_KEY: live}, lk.MEMORY)
def _one(got, kind, key, code=None):
rows = [x for x in got["후보"] if (x["종류"], x["열쇠"]) == (kind, key)]
if code:
rows = [x for x in rows if x["짝"][0]["로직"] == code]
assert len(rows) == 1, rows
return rows[0]
def test_모음_묶기_익명_충돌(client):
before = master_hash()
_seed()
got = alias.gather()
assert got["모은곳"] == {"회사": 2, "사람": 5} and got["반려"] == 0
assert len(got["후보"]) == 4 # 터파기 둘(충돌) · 거푸집 안 씀 · 토사절취 내역맞춤(같은 짝 하나)
same = _one(got, alias.ALIAS, DIG_KEY, "GF000101")
assert (same["회사"], same["사람"], same["충돌"], same["마스터"]) == (1, 2, True, None)
assert _one(got, alias.ALIAS, DIG_KEY, "GF000103")["사람"] == 1
form = _one(got, alias.ALIAS, FORM_KEY)
assert (form["갈래"], form["충돌"], form["사람"]) == ("안 씀", False, 1)
cut = _one(got, alias.BILL, CUT_KEY)
assert (cut["회사"], cut["사람"], cut["확정"], cut["충돌"]) == (2, 2, "earth:4", False)
text = json.dumps(got, ensure_ascii=False)
for secret in ("갑", "을", "병", "가공사.xls", "누가", "처음파일", "주인"):
assert secret not in text # 익명 — 회사 수 · 사람 수만
assert master_hash() == before
def test_승인_마스터_줄_반려_기억(client, tmp_path, monkeypatch):
before = master_hash()
_seed()
newbie = layers.personal_dir(9, 99) # 처음 쓰는 다른 회사 사용자
assert ledger.recall(DIG, newbie) is None
sig = _one(alias.gather(), alias.ALIAS, DIG_KEY, "GF000101")["서명"]
got = alias.approve(alias.ALIAS, DIG_KEY, sig, {"id": 1, "이름": "관리자"})
hit = ledger.recall(DIG, newbie)
assert (hit["층"], hit["짝"][0]["로직"]) == ("system", "GF000101") # 승인 → 모두가 씀
master = ledger.read(client.system)
assert master[DIG_KEY]["처음파일"] is None # 마스터 줄에도 남의 파일 이름 없음
other = _one(got, alias.ALIAS, DIG_KEY) # 승인한 짝 = 마스터와 같음 → 후보에서 빠짐
assert other["짝"][0]["로직"] == "GF000103"
assert (other["마스터"], other["충돌"]) == ("다름", True)
got = alias.reject(alias.ALIAS, DIG_KEY, other["서명"], {"id": 1})
assert got["반려"] == 1 and not [x for x in got["후보"] if x["열쇠"] == DIG_KEY]
assert alias.gather()["반려"] == 1 # 반려 기억 = 다시 모아도 안 보임
with pytest.raises(alias.store.StoreError) as error:
alias.approve(alias.ALIAS, DIG_KEY, other["서명"], {"id": 1})
assert error.value.status == 404
with pytest.raises(alias.store.StoreError):
alias.approve("모름", DIG_KEY, sig, {"id": 1})
# 내역맞춤 승인 → 마스터 기억 → 다른 회사 새 사용자 B09 잇기 기억에 나옴
cut = _one(alias.gather(), alias.BILL, CUT_KEY)
alias.approve(alias.BILL, CUT_KEY, cut["서명"], {"id": 1})
assert [r["열쇠"] for r in alias.master_bill()] == [CUT_KEY]
monkeypatch.setattr(route.access, "folder", lambda got, create=False: tmp_path / got.layer)
user = {"company_id": 9, "user_id": 99, "role": "USER"}
assert route.recalled(user)[CUT_KEY]["확정"] == "earth:4"
assert alias.forget_bill(CUT_KEY) == [] and CUT_KEY not in route.recalled(user)
assert master_hash() == before
def test_길_시스템_관리자만(client):
_seed()
app = FastAPI()
app.include_router(alias.router, dependencies=[Depends(require_system_admin)])
who = {"user_id": 1, "company_id": None, "role": "USER", "name": "갑"}
app.dependency_overrides[verify_session] = lambda: who
web = TestClient(app)
for role in ("USER", "ADMIN"):
who["role"] = role
assert web.get("/api/m01/alias-ledger").status_code == 403
who["role"] = "SYSTEM_ADMIN"
got = web.get("/api/m01/alias-ledger").json()
assert len(got["후보"]) == 4 and got["마스터내역"] == []
form = _one(got, alias.ALIAS, FORM_KEY)
pick = {"종류": alias.ALIAS, "열쇠": FORM_KEY, "서명": form["서명"]}
res = web.post("/api/m01/alias-ledger/approve", json=pick)
assert res.status_code == 200 and len(res.json()["후보"]) == 3
assert ledger.read(client.system)[FORM_KEY]["갈래"] == "안 씀"
assert web.post("/api/m01/alias-ledger/approve", json=pick).status_code == 404
assert web.post("/api/m01/alias-ledger/reject", json={**pick, "서명": "x"}).status_code == 422
gone = web.post("/api/m01/alias-ledger/master-bill/forget", json={"열쇠": CUT_KEY})
assert gone.status_code == 404
main = (ROOT / "main.py").read_text(encoding="utf-8")
at = main.index("M01_MasterData.M01_MasterData_Alias")
assert "dependencies=system_admin_only" in main[at : at + 120]