feat(815-2): M01 별칭 모아 보기 · 제안 → 승인 길
- M01 왼쪽 「별칭 장부」 → 「별칭 모아 보기」 — 별칭이 적힌 줄 전부(인력 · 기계 · 재료 · 유가 · 단가산출 기초) · 수기/승인 표시 · 갈래 거름 · 찾기 · 줄 누르면 그 요소 화면 그 줄 · 별칭 칸 고침 → [저장](그 줄 별칭만) - 위 「승인 대기 n」 — [승인] = 그 요소 별칭에 콤마로 더함(속 콤마 = 큰따옴표 조각) · [반려] = 지움 - 서버 M01_MasterData_Alias 새로 — GET /api/m01/alias · POST save · approve · reject(시스템 관리자) · POST suggest(로그인만 · storage/_shared/별칭제안.json) · 풀이 = master_alias - 옛 /api/m01/alias-ledger(모인 후보 · 마스터 장부 · 반려 기억) 걷음 - M01 저장 — 값이 같은 칸은 파일 글 그대로(화면 왕복 9.0 → 9 막음 · 그 줄 별칭만 바뀜) - 대시보드 카드 이름 「별칭 모아 보기」 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -1,17 +1,22 @@
|
||||
"""M01 별칭 장부 시스템 모으기(PLAN 718-2 · 요구서 내역12 4장 23 · 4-1 확정 6).
|
||||
"""M01 별칭 모아 보기 · 제안 → 승인(PLAN 815-2 · 요구 764 · 765) — 서버 단독.
|
||||
|
||||
모든 회사 · 개인 층의 별칭 장부(`실무자료/맞춤.json` · `M02_Hopyo_Ledger`)와 B09 공사 내역 맞춤
|
||||
기억(`실무자료/내역맞춤.json` · 707-4)을 읽어 후보로 모음 — 읽기만(공유 storage 안 씀).
|
||||
후보 = (종류, 열쇠, 서명) · 서명 = 짝 내용 지문(같은 짝이면 같은 서명) · 회사 수 · 사람 수만
|
||||
(누가 · 처음파일 · 층 주인은 안 내보냄 = 익명) · 충돌 = 같은 열쇠 다른 서명 · 마스터와 다름.
|
||||
승인 = 시스템 관리자만 → 마스터 층에 씀(맞춤 = `ledger.put` 재검사 · 내역맞춤 = 기억 파일) ·
|
||||
반려 = 마스터 층 `실무자료/별칭반려.json` 에 서명 기억(같은 짝은 다시 안 보임) · 자동 반영 없음.
|
||||
별칭 = 요소 · 로직 줄 칸 `별칭` 하나(콤마 여러 개 · 풀이 정본 `master_alias`).
|
||||
GET /api/m01/alias → {줄: 별칭이 적힌 줄 전부(갈래 · 키 · 이름 · 규격 · 별칭 · 조각 ·
|
||||
들어옴 수기/승인), 대기: 승인 대기 제안}
|
||||
POST /api/m01/alias/save {키, 별칭} → 그 줄 `별칭` 만 고침(M01 저장 길 `make.save`)
|
||||
POST /api/m01/alias/approve {id} → 그 요소 `별칭` 에 콤마로 더함 · 제안 = 「승인」 기록
|
||||
POST /api/m01/alias/reject {id} → 제안 지움
|
||||
POST /api/m01/alias/suggest {키, 이름, 규격?, 단위?(갈래는 서버가 키로 정함)} → 제안 하나(로그인만 · 일반 계정도)
|
||||
제안 파일 = `storage/_shared/별칭제안.json` `{제안: {id: {…, 상태 대기|승인}}}`.
|
||||
권한 = `router` 시스템 관리자 · `suggest_router` 로그인만(`main.py`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -20,160 +25,161 @@ 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 M01_MasterData import M01_MasterData_Store as store
|
||||
from M01_MasterData import M01_MasterData_Store_Make as make
|
||||
from M02_MasterTemplete import M02_Template_Layers as layers
|
||||
|
||||
ALIAS, BILL = "맞춤", "내역맞춤"
|
||||
KINDS = (ALIAS, BILL)
|
||||
REJECT = "별칭반려"
|
||||
import master_alias as ma # noqa: E402 (`store` 가 scripts 를 길에 둠)
|
||||
|
||||
router = APIRouter(prefix="/api/m01/alias-ledger", tags=["M01 별칭 장부"])
|
||||
ALIAS = "별칭"
|
||||
FUEL = "재료_유가전력.json"
|
||||
ELEMENTS = ("인력.json", "기계.json", "재료_자재품목.json", FUEL) # 제안을 받는 요소 파일
|
||||
_LOCK = threading.Lock()
|
||||
|
||||
router = APIRouter(prefix="/api/m01/alias", tags=["M01 별칭"])
|
||||
suggest_router = APIRouter(prefix="/api/m01/alias", 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()):
|
||||
def kind_of(file: str, group: str) -> str:
|
||||
if file == FUEL:
|
||||
return "유가"
|
||||
return "단가산출 기초" if group == "로직" else group
|
||||
|
||||
|
||||
# ── 제안 파일 ─────────────────────────────────────────────────────────
|
||||
def suggest_path() -> Path:
|
||||
return layers.storage_root() / "_shared" / "별칭제안.json"
|
||||
|
||||
|
||||
def _book() -> dict[str, dict[str, Any]]:
|
||||
try:
|
||||
got = json.loads(suggest_path().read_text(encoding="utf-8")).get("제안")
|
||||
except (OSError, ValueError, AttributeError):
|
||||
return {}
|
||||
return got if isinstance(got, dict) else {}
|
||||
|
||||
|
||||
def _keep(book: dict[str, dict[str, Any]]) -> None:
|
||||
path = suggest_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = path.with_suffix(".tmp")
|
||||
tmp.write_text(json.dumps({"제안": book}, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
os.replace(tmp, path)
|
||||
|
||||
|
||||
# ── 줄 찾기 · 고치기 ───────────────────────────────────────────────────
|
||||
def _find(key: str, files: tuple[str, ...] | None = None) -> tuple[str, dict[str, Any]]:
|
||||
loaded, _ = store.loaded()
|
||||
for name, data in loaded.items():
|
||||
if files and name not in files:
|
||||
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
|
||||
for row in data.get("줄") or []:
|
||||
if str(row.get("키")) == key:
|
||||
return name, row
|
||||
raise store.StoreError(404, f"없는 키 「{key}」")
|
||||
|
||||
|
||||
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 set_alias(key: str, text: str) -> dict[str, Any]:
|
||||
"""그 줄 `별칭` 만 바꿔 M01 저장 길로(빈 글 = 칸 없음 · 별칭만이라 정본 개정 확인 안 물음)."""
|
||||
file, _ = _find(key)
|
||||
data, version = store.read(file)
|
||||
row = dict(next(r for r in data["줄"] if str(r.get("키")) == key))
|
||||
row[ALIAS] = text
|
||||
change = {"op": "edit", "key": key, "row": row}
|
||||
make.save([{"file": file, "version": version, "changes": [change]}], True)
|
||||
return board()
|
||||
|
||||
|
||||
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 _quoted(piece: str) -> str:
|
||||
return f'"{piece}"' if any(s in piece for s in (",", "，")) else piece
|
||||
|
||||
|
||||
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 _joined(text: Any, piece: str) -> str:
|
||||
"""별칭 글 끝에 한 조각 — 이미 있으면(정규화 같음) 그대로."""
|
||||
raw = str(text or "").strip()
|
||||
if ma.norm(piece) in {ma.norm(x) for x in ma.names(raw)}:
|
||||
return raw
|
||||
return f"{raw}, {_quoted(piece)}" if raw else _quoted(piece)
|
||||
|
||||
|
||||
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 board() -> dict[str, Any]:
|
||||
book = _book()
|
||||
approved = {
|
||||
(str(v.get("키")), ma.norm(v.get(ALIAS))) for v in book.values() if v.get("상태") == "승인"
|
||||
}
|
||||
rows = []
|
||||
loaded, _ = store.loaded()
|
||||
for file, data in loaded.items():
|
||||
kind = kind_of(file, str(data.get("그룹") or ""))
|
||||
for row in data.get("줄") or []:
|
||||
if not str(row.get(ALIAS) or "").strip():
|
||||
continue
|
||||
key = str(row.get("키"))
|
||||
pieces = [
|
||||
{"글": p, "승인": (key, ma.norm(p)) in approved} for p in ma.names(row[ALIAS])
|
||||
]
|
||||
rows.append(
|
||||
{
|
||||
"갈래": kind,
|
||||
"파일": file,
|
||||
"키": key,
|
||||
"이름": str(row.get("이름") or ""),
|
||||
"규격": str(row.get("규격") or ""),
|
||||
ALIAS: str(row[ALIAS]),
|
||||
"조각": pieces,
|
||||
"들어옴": "승인" if any(p["승인"] for p in pieces) else "수기",
|
||||
}
|
||||
)
|
||||
waiting = [{"id": k, **v} for k, v in book.items() if v.get("상태") == "대기"]
|
||||
waiting.sort(key=lambda x: str(x.get("때") or ""))
|
||||
return {"줄": rows, "대기": waiting}
|
||||
|
||||
|
||||
def rejected() -> dict[str, dict[str, Any]]:
|
||||
return hopyo_store.memory(store.FOLDER, REJECT)
|
||||
# ── 제안 · 승인 · 반려 ────────────────────────────────────────────────
|
||||
def suggest(key: str, name: str, spec: str, unit: str, who: Any) -> dict[str, Any]:
|
||||
file, row = _find(key, ELEMENTS)
|
||||
piece = " ".join(x for x in (name.strip(), spec.strip()) if x)
|
||||
if not piece:
|
||||
raise store.StoreError(400, "가져온 이름이 비었음")
|
||||
if _joined(row.get(ALIAS), piece) == str(row.get(ALIAS) or "").strip():
|
||||
return {"이미": True} # 이미 그 별칭이 있음
|
||||
sid = hashlib.sha256(f"{key}|{ma.norm(piece)}".encode()).hexdigest()[:16]
|
||||
with _LOCK:
|
||||
book = _book()
|
||||
if book.get(sid, {}).get("상태") != "승인":
|
||||
book[sid] = {
|
||||
"갈래": kind_of(file, str(store.loaded()[0][file].get("그룹") or "")),
|
||||
"키": key,
|
||||
"이름": name.strip(),
|
||||
"규격": spec.strip(),
|
||||
"단위": unit.strip(),
|
||||
ALIAS: piece,
|
||||
"상태": "대기",
|
||||
"누가": who,
|
||||
"때": utc_now_text(),
|
||||
}
|
||||
_keep(book)
|
||||
return {"id": sid}
|
||||
|
||||
|
||||
def _reject_id(kind: str, full: str, sig: str) -> str:
|
||||
return f"{kind}|{sig}|{full}"
|
||||
def decide(sid: str, ok: bool, who: Any) -> dict[str, Any]:
|
||||
with _LOCK:
|
||||
book = _book()
|
||||
one = book.get(sid)
|
||||
if not one or one.get("상태") != "대기":
|
||||
raise store.StoreError(404, "대기 중인 제안이 없음(이미 승인 · 반려)")
|
||||
if ok:
|
||||
_, row = _find(str(one["키"]), ELEMENTS)
|
||||
set_alias(str(one["키"]), _joined(row.get(ALIAS), str(one[ALIAS])))
|
||||
one.update(상태="승인", 승인={"누가": who, "때": utc_now_text()})
|
||||
else:
|
||||
del book[sid]
|
||||
_keep(book)
|
||||
return board()
|
||||
|
||||
|
||||
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")}
|
||||
|
||||
@@ -185,31 +191,44 @@ def _call(fn, *args):
|
||||
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)
|
||||
class AliasBody(BaseModel):
|
||||
키: str = Field(min_length=1, max_length=20)
|
||||
별칭: str = Field(max_length=2000)
|
||||
|
||||
|
||||
class Decide(BaseModel):
|
||||
id: str = Field(min_length=16, max_length=16)
|
||||
|
||||
|
||||
class Suggest(BaseModel):
|
||||
갈래: str = Field(default="", max_length=20)
|
||||
키: str = Field(min_length=1, max_length=20)
|
||||
이름: str = Field(min_length=1, max_length=200)
|
||||
규격: str = Field(default="", max_length=200)
|
||||
단위: str = Field(default="", max_length=40)
|
||||
|
||||
|
||||
@router.get("")
|
||||
def candidates() -> dict[str, Any]:
|
||||
return {**gather(), "마스터내역": master_bill()}
|
||||
def get_board() -> dict[str, Any]:
|
||||
return board()
|
||||
|
||||
|
||||
@router.post("/save")
|
||||
def post_save(body: AliasBody) -> dict[str, Any]:
|
||||
return _call(set_alias, body.키, body.별칭)
|
||||
|
||||
|
||||
@router.post("/approve")
|
||||
def approve_one(body: Pick, session: dict = Depends(verify_session)) -> dict[str, Any]:
|
||||
return _call(approve, body.종류, body.열쇠, body.서명, _who(session))
|
||||
def post_approve(body: Decide, session: dict = Depends(verify_session)) -> dict[str, Any]:
|
||||
return _call(decide, body.id, True, _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))
|
||||
def post_reject(body: Decide, session: dict = Depends(verify_session)) -> dict[str, Any]:
|
||||
return _call(decide, body.id, False, _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.열쇠)}
|
||||
@suggest_router.post("/suggest")
|
||||
def post_suggest(body: Suggest, session: dict = Depends(verify_session)) -> dict[str, Any]:
|
||||
args = (body.키, body.이름, body.규격, body.단위, _who(session))
|
||||
return _call(suggest, *args)
|
||||
|
||||
Reference in New Issue
Block a user