Aislo/M02_MasterTemplete/M02_Library_Same_Router.py

"""M02 서고 [중복 찾기] · 합치기 · 되돌림(PLAN 733-3 · 요구서_572 3장 4 · 분석 D-3) —
단가산출(`sangeun` SK) · 일위대가(`precombo` SC).

GET  /layers/{층}/{kind}-same — 이 층 + 위 층(회사 · 마스터) 같은 지문 묶음(이 층 문서가 든
     묶음만) · 문서마다 층 · 키 · 이름 · 구분 · 쓰는 곳 수 · 꼬리표 「내용 같음」 ·
     SC 「수량만 다름」(`본뜬.실무수량` 다름) · 합친 것 목록(되돌림)
POST …/merge {남김, 지움[], 판} — 서버가 지문 다시 셈(다르면 409) · 쓰는 곳 참조를 남김으로
     (쓰기 권한 닿는 곳만 · `store.write` 정본 다시 셈) · 남은 쓰는 곳이 있는 문서 = 안 지움
     「남김 — 쓰는 곳 N」 · 나머지 = `{층}/_합친것/<키>.json` 으로 옮김 · 나무 자리 걷음
POST …/undo {키} — `_합친것/` 에서 되돌림 · 옮긴 참조도 되돌림 · 나무 지움 기록(735-1 `지움` 칸)이면
     `M02_Library_Delete.undo`(문서 · 자리 · 고친 문서 앞판)
POST /layers/{층}/{kind}-tree/delete-check {대상} — 나무 지우기 미리 보기(735-1) · 지울 문서 ·
     쓰는 줄
지움 = 이 층 문서만 · 남김 = 이 층 또는 위 층(위 층과 같은 아래 층 문서 = 남김 = 위 층 키) ·
판 = 이 층 서고 나무 판(`{종류}_폴더.json`) · 권한 = 그 층 쓰기 · 프로젝트 사본 = 굳힌 값이라
안 건드림 · 지문 = `M02_Library_Same`(733-1) 읽기만.
⚠ 등록은 `main.py`(로그인만).
"""

from __future__ import annotations

import json
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_Labor_Ref as refs
from M02_MasterTemplete import M02_Library_Delete as deleting
from M02_MasterTemplete import M02_Library_Same as same
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Precombo_Folders as pre_folders
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_MasterTemplete_Router_Items import Layer, Session, _root

router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Library Same"])

Kind = Literal["precombo", "sangeun"]
GONE = deleting.GONE
USER_KINDS = deleting.USER_KINDS
_ORDER = {"system": 0, "company": 1, "personal": 2}


class MergeBody(BaseModel):
    """`남김` = {층, 주인, 키}(이 층 · 위 층) · `지움` = 이 층 키들 · `판` = 이 층 서고 나무 판."""

    남김: dict[str, Any]
    지움: list[str]
    판: str


class UndoBody(BaseModel):
    키: str


class CheckBody(BaseModel):
    """나무 지우기 미리 보기 — 나무 고침 `대상` 꼴 `[{경로, 키?}]`."""

    대상: list[dict[str, Any]]


def _bad(status: int, text: Any) -> store.StoreError:
    return store.StoreError(status, text)


def _ref(at: refs.Here, key: str) -> refs.Ref:
    return {"층": at[0], "주인": at[1], "키": key}


def _docs(at: refs.Here, kind: str) -> list[tuple[str, dict[str, Any]]]:
    folder = refs.folder_of(*at) / kind
    out = []
    for path in sorted(folder.glob("[!._]*.json")) if folder.is_dir() else []:
        try:
            doc = json.loads(path.read_bytes())
        except (OSError, ValueError):
            continue
        if isinstance(doc, dict):
            out.append((path.stem, doc))
    return out


# ── 쓰는 곳 ───────────────────────────────────────────────────────────
def _user_index() -> dict[str, list[tuple[Path, str, str]]]:
    """모든 층(프로젝트 뺌) 훑기 한 번 — {ref_id: [(파일, 종류, 「어디 이름」)]}."""
    out: dict[str, list[tuple[Path, str, str]]] = {}
    for kind, what in USER_KINDS:
        for where, path in layers.layer_files(kind):
            try:
                doc = json.loads(path.read_bytes())
            except (OSError, ValueError):
                continue
            name = doc.get("일위대가번") or doc.get("이름") or path.stem
            for rid in {refs.ref_id(one) for one in refs.refs_in(doc)}:
                out.setdefault(rid, []).append((path, kind, f"{what} {where} 「{name}」"))
    return out


def _can_write(session: Session, at: refs.Here) -> bool:
    return deleting.can_write(session, at)


def _retarget(doc: dict[str, Any], olds: set[str], new: refs.Ref) -> dict[str, Any]:
    """문서의 `olds` 참조를 모두 `new` 로(규격 열 수량 열쇠까지 · `M02_Labor_Ref` 한 벌)."""
    done: dict[str, refs.Ref] = {}

    def move(ref: refs.Ref, _depth: int) -> refs.Ref:
        if refs.ref_id(ref) in olds:
            done[refs.ref_id(ref)] = new
            return new
        return ref

    return refs._spec_moved(refs._move_doc(doc, move, 1), done)


def _rewrite(path: Path, kind: str, olds: set[str], new: refs.Ref) -> None:
    """쓰는 곳 한 파일 — 참조를 바꿔 `store.write`(서버 정본 다시 셈 · 판 검사)."""
    raw = path.read_bytes()
    doc = _retarget(json.loads(raw), olds, new)
    with store.at(path.parent.parent):
        store.write(kind, path.stem, store.version_of(raw), doc)


# ── 목록 ──────────────────────────────────────────────────────────────
def _qty(doc: dict[str, Any]) -> str:
    return json.dumps((doc.get("본뜬") or {}).get("실무수량") or {}, sort_keys=True)


def _row(at: refs.Here, key: str, doc: dict[str, Any], used: int, mine: refs.Here) -> dict:
    return {
        "층": at[0],
        "주인": at[1],
        "키": key,
        "이름": doc.get("이름") or key,
        "구분": doc.get("구분") or "",
        "상세구분": doc.get("상세구분") or "",
        "쓰는곳수": used,
        "위층": at != mine,
    }


def _gone_rows(folder: Path, kind: str) -> list[dict[str, Any]]:
    out = []
    for path in sorted((folder / GONE).glob("*.json")):
        try:
            rec = json.loads(path.read_bytes())
        except (OSError, ValueError):
            continue
        if rec.get("종류") != kind:
            continue
        if "지움" in rec:  # 나무 지움(735-1) — 남김 없음 · 이름 = 지운 문서들
            names = [(d.get("문서") or {}).get("이름") or d.get("키") for d in rec["지움"]]
            row = {"이름": " · ".join(map(str, names)), "남김": None, "고친": rec.get("고친곳")}
        else:
            doc = rec.get("문서") or {}
            row = {"이름": doc.get("이름") or rec.get("키"), "남김": rec.get("남김"), "고친": None}
        out.append(
            {
                "키": path.stem,
                "이름": row["이름"],
                "남김": row["남김"],
                "때": rec.get("때"),
                "옮긴곳수": len(row["고친"] or rec.get("옮긴곳") or []),
            }
        )
    return out


def find(folder: Path, kind: str) -> dict[str, Any]:
    """이 층 + 위 층 같은 지문 묶음 — 이 층 문서가 든 묶음 · 둘 이상만."""
    mine = refs.where(folder)
    index = _user_index()
    groups: dict[str, list[tuple[refs.Here, str, dict[str, Any]]]] = {}
    marks = same.Same()  # 한 요청 안 참조 지문 메모
    for at in [mine, *same.above(mine)]:
        for key, doc in _docs(at, kind):
            if (mark := marks.finger(kind, doc)) is not None:
                groups.setdefault(mark, []).append((at, key, doc))
    out = []
    for mark, docs in groups.items():
        if len(docs) < 2 or all(at != mine for at, _, _ in docs):
            continue
        docs.sort(key=lambda d: (_ORDER[d[0][0]], d[1]))
        rows = [
            _row(at, key, doc, len(index.get(refs.ref_id(_ref(at, key)), [])), mine)
            for at, key, doc in docs
        ]
        qty = kind == "precombo" and len({_qty(doc) for _, _, doc in docs}) > 1
        out.append({"지문": mark, "꼬리표": "수량만 다름" if qty else "내용 같음", "문서": rows})
    out.sort(key=lambda g: (g["꼬리표"] != "내용 같음", g["문서"][0]["이름"]))
    with store.at(folder):
        판 = pre_folders.view(kind)["판"]
    return {"판": 판, "묶음": out, "합친것": _gone_rows(folder, kind)}


# ── 합치기 ────────────────────────────────────────────────────────────
def merge(session: Session, folder: Path, kind: str, body: MergeBody) -> dict[str, Any]:
    mine = refs.where(folder)
    raw = body.남김 or {}
    keep_at: refs.Here = (str(raw.get("층") or ""), raw.get("주인") or None)
    keep = _ref(keep_at, str(raw.get("키") or ""))
    if keep_at not in [mine, *same.above(mine)]:
        raise _bad(400, "남김은 이 층 또는 위 층 문서")
    if refs.kind_of(keep["키"]) != kind or (keep_doc := refs.read_ref(keep)) is None:
        raise _bad(404, f"없는 {refs.TITLE[kind]} 「{keep['키']}」")
    gone = list(dict.fromkeys(str(k) for k in body.지움))
    if not gone or (keep_at == mine and keep["키"] in gone):
        raise _bad(400, "지울 문서를 고름 — 남김과 달라야 함")
    have = dict(_docs(mine, kind))
    if miss := [k for k in gone if k not in have]:
        raise _bad(404, f"이 층에 없는 문서 「{' · '.join(miss)}」")
    with store.at(folder):
        tree_doc, 판 = pre_folders._read(kind)
    if 판 != body.판:
        raise _bad(409, {"stale": [pre_folders.file_of(kind)], "판": 판})
    marks = same.Same()
    mark = marks.finger(kind, keep_doc)
    if mark is None or any(marks.finger(kind, have[k]) != mark for k in gone):
        raise _bad(409, "내용이 그새 달라짐 — [중복 찾기]를 다시 누름")

    rids = {refs.ref_id(_ref(mine, k)): k for k in gone}
    by_file: dict[Path, tuple[str, set[str]]] = {}
    labels: dict[Path, str] = {}
    index = _user_index()
    for rid in rids:
        for path, ukind, label in index.get(rid, []):
            by_file.setdefault(path, (ukind, set()))[1].add(rid)
            labels[path] = label
    moved: dict[str, list[dict[str, Any]]] = {k: [] for k in gone}
    left: dict[str, list[str]] = {k: [] for k in gone}
    for path, (ukind, olds) in by_file.items():
        at = refs.where(path.parent.parent)
        why = ""
        if not (_can_write(session, at) and refs.readable(at, keep)):
            why = "권한 밖"
        else:
            try:
                _rewrite(path, ukind, olds, keep)
            except store.StoreError as error:  # 같은 참조 두 번 등 — 그 쓰는 곳은 그대로 둠
                why = str(error.detail)
        for rid in olds:
            if why:
                left[rids[rid]].append(f"{labels[path]} — {why}")
            else:
                moved[rids[rid]].append(
                    {"층": at[0], "주인": at[1], "종류": ukind, "이름": path.stem}
                )

    places = dict(tree_doc.get("자리") or {})
    done, kept = [], []
    when = datetime.now().isoformat(timespec="seconds")
    # ponytail: 여러 파일을 차례로 씀(잠금 없음) — 동시에 같은 층을 합치면 앞 것이 이김
    for key in gone:
        if left[key]:
            kept.append({"키": key, "쓰는곳": left[key]})
            continue
        path = folder / kind / f"{key}.json"
        record = {
            "키": key,
            "종류": kind,
            "문서": json.loads(path.read_bytes()),
            "자리": places.pop(key, None),
            "남김": keep,
            "옮긴곳": moved[key],
            "때": when,
            "누가": session.get("name"),
        }
        atomic_write_json(folder / GONE / f"{key}.json", record)
        path.unlink()
        done.append({"키": key, "옮긴곳수": len(moved[key])})
    if done:
        atomic_write_json(folder / pre_folders.file_of(kind), {**tree_doc, "자리": places})
    return {"합침": done, "남김": kept, **find(folder, kind)}


def undo(session: Session, folder: Path, kind: str, key: str) -> dict[str, Any]:
    """`_합친것/<키>` 되돌림 — 문서 · 나무 자리 · 옮긴 참조(아직 남김을 가리키고 쓸 수 있는 곳)."""
    mine = refs.where(folder)
    rec_path = folder / GONE / f"{key}.json"
    try:
        rec = json.loads(rec_path.read_bytes())
    except (OSError, ValueError):
        raise _bad(404, f"합친 것에 없음 「{key}」") from None
    if rec.get("종류") != kind:
        raise _bad(404, f"합친 것에 없음 「{key}」")
    if "지움" in rec:
        got = deleting.undo(session, folder, kind, rec)
        rec_path.unlink()
        return {"되돌림": key, **got, **find(folder, kind)}
    path = folder / kind / f"{key}.json"
    if path.exists():
        raise _bad(409, f"「{key}」 키를 그새 다른 문서가 씀 — 되돌릴 수 없음")
    atomic_write_json(path, rec["문서"])
    with store.at(folder):
        tree_doc, _ = pre_folders._read(kind)
    if rec.get("자리"):
        places = {**(tree_doc.get("자리") or {}), key: rec["자리"]}
        atomic_write_json(folder / pre_folders.file_of(kind), {**tree_doc, "자리": places})
    keep, back, miss = rec.get("남김") or {}, 0, []
    new = _ref(mine, key)
    # ponytail: 옮긴 뒤 그 쓰는 곳이 남김을 다른 줄에서 따로 담았으면 그 줄까지 되돌림 — 드문 꼴
    for one in rec.get("옮긴곳") or []:
        at = (one.get("층"), one.get("주인"))
        user = refs.folder_of(*at) / str(one.get("종류")) / f"{one.get('이름')}.json"
        try:
            if not _can_write(session, at) or not user.is_file():
                raise store.StoreError(403, "권한 밖 · 없음")
            _rewrite(user, str(one.get("종류")), {refs.ref_id(keep)}, new)
            back += 1
        except store.StoreError as error:
            miss.append(f"{one.get('종류')} {one.get('이름')} — {error.detail}")
    rec_path.unlink()
    return {"되돌림": key, "되돌린곳수": back, "못한곳": miss, **find(folder, kind)}


def _run(fn, *args):
    try:
        return fn(*args)
    except store.StoreError as e:
        raise HTTPException(status_code=e.status, detail=e.detail) from e


@router.get("/layers/{layer}/{kind}-same")
def library_same(
    layer: Layer, kind: Kind, owner: str | None = None, session: Session = Depends(verify_session)
) -> dict:
    """[중복 찾기] — 이 층 + 위 층 같은 내용 묶음 · 합친 것 목록(읽기 = 그 층 쓰기 권한)."""
    return _run(find, _root(session, layer, owner, write=True), kind)


@router.post("/layers/{layer}/{kind}-same/merge")
def library_same_merge(
    layer: Layer,
    kind: Kind,
    body: MergeBody,
    owner: str | None = None,
    session: Session = Depends(verify_session),
) -> dict:
    """합치기 — 쓰는 곳 참조 → 남김 · 지운 문서 `_합친것/` · 판 · 지문 다르면 409."""
    return _run(merge, session, _root(session, layer, owner, write=True), kind, body)


def check(session: Session, folder: Path, kind: str, picks: list[dict[str, Any]]) -> dict:
    """나무 지우기 미리 보기 — 나무 고침을 셈만(저장 안 함) → 지울 문서 · 쓰는 줄."""
    with store.at(folder):
        tree_doc, _ = pre_folders._read(kind)
        try:
            _, note = pre_folders.apply_many(
                tree_doc, {"op": "delete", "대상": picks}, pre_folders._keys(kind)
            )
        except pre_folders.tree.TreeError as e:
            raise _bad(e.status, e.detail) from e
        gone = note.get("지운문서") or []
        return {"문서": gone, "쓰는곳": deleting.users(session, kind, gone)}


@router.post("/layers/{layer}/{kind}-tree/delete-check")
def library_delete_check(
    layer: Layer,
    kind: Kind,
    body: CheckBody,
    owner: str | None = None,
    session: Session = Depends(verify_session),
) -> dict:
    """나무 지우기 미리 보기(735-1) — `{문서, 쓰는곳}`(`M02_Library_Delete.users` 줄)."""
    return _run(check, session, _root(session, layer, owner, write=True), kind, body.대상)


@router.post("/layers/{layer}/{kind}-same/undo")
def library_same_undo(
    layer: Layer,
    kind: Kind,
    body: UndoBody,
    owner: str | None = None,
    session: Session = Depends(verify_session),
) -> dict:
    """되돌림 — `_합친것/<키>` 문서 · 나무 자리 · 옮긴 참조."""
    return _run(undo, session, _root(session, layer, owner, write=True), kind, body.키)