Aislo/M01_MasterData/M01_MasterData_Alias.py

"""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.열쇠)}