Merge remote-tracking branch 'origin/sub_laptop_5' into sub_laptop_6

B09_Estimation/B09_Estimation_Bill_Link.py

@@ -15,10 +15,15 @@
     안 줄이 그 아래(저절로 묶음 · 금액 = 안 줄 합) / price = 안 줄 금액 합 ÷ 개수(기본 = B08 수량)
     = 1 단위 고정 단가 · 안 줄 지움
     (B08 수량 × 단가 · 끝수 비목마다 1 원 안).
+    이은 자동 줄 `was` = 되돌릴 것 {rows: [불러온 줄 · price 안 줄], old: 자동 줄 덮개,
+    prev · next: 그때 같은 부모 앞 · 뒤 줄 키} · 수량 갈래 = 줄마다(`modes`) · 없으면 `mode`
+풀기(`unlink`) — `was` 대로 불러온 줄 · 자동 줄 제자리(맞춤 기억은 그대로 · 잊기 = 부르는 쪽).
+판(`board`) — 남은 불러온 줄 · 빈 자리(이을 수 있는 자동 줄) · 이은 줄.
 """
 
 from __future__ import annotations
 
+import copy
 import re
 from typing import Any
 
@@ -198,11 +203,14 @@ def link(
     pairs: list[tuple[str, str]],
     mode: str = "take",
     counts: dict[str, float] | None = None,
+    modes: dict[str, str] | None = None,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """(새 사용자 값, 알림 {이음, 묶음, 지움, 못함 [{key, 까닭}], 기억 {열쇠: 확정}}) —
     `rows` = 지금 셈 줄 ·
-    `counts` = 묶음 개수 {불러온 묶음 키: 수}(price 갈래 · 없으면 B08 수량)."""
-    if mode not in MODES:
+    `counts` = 묶음 개수 {불러온 묶음 키: 수}(price 갈래 · 없으면 B08 수량) ·
+    `modes` = 줄마다 수량 갈래 {불러온 줄 키: take | price}."""
+    modes = modes or {}
+    if any(m not in MODES for m in (mode, *modes.values())):
         raise ValueError(f"수량 갈래는 {', '.join(MODES)} 중 하나")
     counts = counts or {}
     by_key = {r["key"]: r for r in rows}
@@ -211,6 +219,7 @@ def link(
     bill = freeze(user, rows)
     report: dict[str, Any] = {"이음": 0, "묶음": 0, "지움": 0, "못함": [], "기억": {}}
     for line, target in pairs:
+        mode_ = modes.get(line, mode)
         entries = bill["rows"]
         at = next((n for n, e in enumerate(entries) if e["key"] == line), -1)
         row, goal = by_key.get(line), targets.get(target)
@@ -221,32 +230,37 @@ def link(
             else "묶음 ↔ 구조물 줄만" if (row["kind"] != "item") != target.startswith("st:")
             else ""
         )  # fmt: skip
-        if not why and row["kind"] != "item" and mode == "price":
+        if not why and row["kind"] != "item" and mode_ == "price":
             count = counts.get(line, goal.get("auto_qty", goal.get("qty")))
             why = "" if count else "개수 없음(B08 수량 없음) — 넘겨받기로"
         if why:
             report["못함"].append({"key": line, "까닭": why})
             continue
-        old = next((e for e in entries if e["key"] == target), None)
-        if old is not None:
-            entries.remove(old)
+        old_at = next((n for n, e in enumerate(entries) if e["key"] == target), -1)
+        old = entries.pop(old_at) if old_at >= 0 else None
+        sibs = [e["key"] for e in entries if old and e.get("parent") == old.get("parent")]
+        cut = sum(1 for e in entries[:old_at] if old and e.get("parent") == old.get("parent"))
+        was = {"old": copy.deepcopy(old), "prev": sibs[:cut], "next": sibs[cut:]}
         at = next(n for n, e in enumerate(entries) if e["key"] == line)
         hand = entries.pop(at)
-        auto = {k: v for k, v in (old or {"key": target}).items() if k not in ("price", "src_qty")}
+        was["rows"] = [copy.deepcopy(hand)]
+        skip = ("price", "src_qty", "was")
+        auto = {k: v for k, v in (old or {"key": target}).items() if k not in skip}
         if row["kind"] == "item":
             auto.update({"ref": hand.get("ref"), "src_qty": hand.get("src_qty", hand.get("qty"))})
             if hand.get("price"):
                 auto["price"] = hand["price"]
-            if mode == "take" or goal.get("auto_qty", goal.get("qty")) is None:
+            if mode_ == "take" or goal.get("auto_qty", goal.get("qty")) is None:
                 auto["qty"] = hand.get("qty")
             report["이음"] += 1
-        elif mode == "take":
+        elif mode_ == "take":
             for e in entries:  # 안 줄 = 구조물 줄 아래 · 금액 = 안 줄 합
                 if e.get("parent") == line:
                     e["parent"] = target
             report["묶음"] += 1
         else:
             inner = _descendants(entries, line)
+            was["rows"] += [copy.deepcopy(e) for e in entries if e["key"] in inner]
             items = [by_key[k] for k in inner if by_key.get(k, {}).get("kind") == "item"]
             auto.update(
                 {
@@ -258,7 +272,73 @@ def link(
             bill["rows"] = entries = [e for e in entries if e["key"] not in inner]
             at = min(at, len(entries))
             report.update(묶음=report["묶음"] + 1, 지움=report["지움"] + len(inner))
+        auto["was"] = was
         _place(bill, at, auto, hand.get("parent"))
         used.add(target)
         report["기억"][key_of(row)] = {"확정": target, "이름": goal["name"]}
     return {**user, "bill": bill}, report
+
+
+def unlink(
+    user: dict[str, Any], rows: list[dict[str, Any]], target: str
+) -> tuple[dict[str, Any], str]:
+    """(새 사용자 값, 못한 까닭 · 됨 = "") — 이은 자동 줄 `was` 대로 불러온 줄 · 자동 줄 되돌림."""
+    bill = freeze(user, rows)
+    entries = bill["rows"]
+    at = next((n for n, e in enumerate(entries) if e["key"] == target), -1)
+    was = entries[at].get("was") if at >= 0 else None
+    back = was.get("rows") if isinstance(was, dict) else None
+    if not back:
+        return user, "이은 줄이 아님"
+    entries.pop(at)
+    head = back[0]["key"]
+    for e in entries:  # take 묶음 안 줄 = 불러온 묶음 아래로
+        if e.get("parent") == target:
+            e["parent"] = head
+    entries[at:at] = copy.deepcopy(back)
+    old = was.get("old")
+    if old:  # 자동 줄 = 그때 뒤 줄 앞 · 없으면 앞 줄 뒤 · 둘 다 없음 = 셈이 기본 부모 끝
+        where = {e["key"]: n for n, e in enumerate(entries)}
+        after = [where[k] + 1 for k in was.get("prev") or [] if k in where]
+        spot = next((where[k] for k in was.get("next") or [] if k in where), None)
+        spot = spot if spot is not None else (after[-1] if after else None)
+        if spot is not None:
+            entries.insert(spot, old)
+    return {**user, "bill": bill}, ""
+
+
+def _under(rows: list[dict[str, Any]], key: str) -> bool:
+    """불러온 줄이 이은 자동 줄(구조물 take 묶음) 아래인지."""
+    by_key = {r["key"]: r for r in rows}
+    up = by_key.get(key, {}).get("parent")
+    while up:
+        if not str(up).startswith("hand:"):
+            return up in by_key and str(up).split(":")[0] in HEADS
+        up = by_key.get(up, {}).get("parent")
+    return False
+
+
+def board(user: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]:
+    """잇기 창 판 — 남은 불러온 줄(구조물 묶음에 딸린 줄 뺌) · 빈 자리 · 이은 줄."""
+
+    def pick(r: dict[str, Any]) -> dict[str, Any]:
+        return {f: r.get(f) for f in ("key", "kind", "name", "spec", "unit")}
+
+    used = _linked(rows)
+    entries = {e["key"]: e for e in ((user.get("bill") or {}).get("rows") or [])}
+    by_key = {r["key"]: r for r in rows}
+    linked = []
+    for key, e in entries.items():
+        was = e.get("was") if isinstance(e.get("was"), dict) else None
+        if was and was.get("rows") and key in by_key:
+            hand = was["rows"][0]
+            linked.append({**pick(by_key[key]), "from": pick(hand), "memo": key_of(hand)})
+    return {
+        "남은": [
+            pick(r) for r in rows if r["key"].startswith("hand:x") and not _under(rows, r["key"])
+        ],
+        "자리": [
+            {**pick(r), "head": k.split(":")[0]} for k, r in _targets(rows).items() if k not in used
+        ],
+        "이은": linked,
+    }

B09_Estimation/B09_Estimation_Router.py

@@ -88,6 +88,7 @@ from B09_Estimation.B09_Estimation_Labor_Source import daega_inputs
 from B09_Estimation.B09_Estimation_Linked import linked_resource_keys
 from B09_Estimation.B09_Estimation_Rates import rate_candidates, rate_table
 from B09_Estimation.B09_Estimation_Router_Import import router as import_router
+from B09_Estimation.B09_Estimation_Router_Link import router as link_router
 from B09_Estimation.B09_Estimation_Router_Quotes import router as quotes_router
 from B09_Estimation.B09_Estimation_Summary import compute_summary, hopyo_lines, used_materials
 from common_util.common_util_auth import verify_session
@@ -589,6 +590,7 @@ async def past_import(
 # 166-4 견적서 파일 길 · 707-1~3 경쟁사 결과물 불러오기 길 — 같은 머리 · 같은 권한
 router.include_router(quotes_router)
 router.include_router(import_router)
+router.include_router(link_router)  # 707-4 내역 줄 ↔ 우리 공종 잇기
 
 
 # 302 초안 길(draft/*_Router.py 의 router · 폴더째 지우면 길도 없음) — 같은 머리 · 같은 권한

B09_Estimation/B09_Estimation_Router_Link.py

@@ -1,14 +1,18 @@
 """B09 내역 줄 ↔ 우리 공종 잇기 길(PLAN 707-4) — 저장 안 함(화면 캐시로 → [저장] 때 서버 정본).
 
     POST /api/projects/{id}/estimation/link/suggest   {user?}
-        → {줄: 제안(`B09_Estimation_Bill_Link.suggest`) · 저절로: 지난번 이음 · 하나 수}
+        → {줄: 제안(`B09_Estimation_Bill_Link.suggest` · 구조물 묶음에 딸린 줄 뺌) ·
+           저절로: 지난번 이음 · 하나 수 · 남은 · 자리 · 이은(`board`)}
     POST /api/projects/{id}/estimation/link/apply
-        {pairs [{line, target}], auto, mode, counts?, user?}
+        {pairs [{line, target, mode?}], auto, mode, counts?, user?}
         잇기(`link`) — `auto` = 지난번 이음 · 하나도 더함 →
         {user(새 캐시), 알림, payload(그 캐시로 셈)} ·
         이은 짝 = 맞춤 기억에 씀(회사 층 = 회사 관리자 · 아니면 개인 층)
+    POST …/link/undo {target, user?} → {user, 알림, payload} — 잇기 풀기(맞춤 기억 그대로)
+    POST …/link/forget {key, user?} → {잊음, 기억층, 남음} — 이 짝 잊기(불러온 줄 · 이은 줄 키 →
+        맞춤 열쇠 · 쓸 수 있는 층에서 지움 · 남음 = 못 지운 위 층 기억)
 맞춤 기억 = 층 `실무자료/내역맞춤.json`(실무자료 맞춤 꼴) · 읽기 = 회사 + 개인(개인이 이김).
-끼워 넣기 = `B09_Estimation_Router`(같은 머리 · 같은 권한) — 화면 고리 일감 몫.
+끼워 넣기 = `B09_Estimation_Router`(같은 머리 · 같은 권한) · 화면 = `B09_Estimation_UI_Link`.
 """
 
 from __future__ import annotations
@@ -73,6 +77,17 @@ class SuggestBody(BaseModel):
 class Pair(BaseModel):
     line: str = Field(min_length=1, max_length=80)
     target: str = Field(min_length=1, max_length=200)
+    mode: str | None = None
+
+
+class UndoBody(BaseModel):
+    target: str = Field(min_length=1, max_length=200)
+    user: dict[str, Any] | None = None
+
+
+class ForgetBody(BaseModel):
+    key: str = Field(min_length=1, max_length=200)
+    user: dict[str, Any] | None = None
 
 
 class ApplyBody(BaseModel):
@@ -96,8 +111,11 @@ async def suggest(
     root = await _project_root(project_id)
     draft = _draft(body.user) or await asyncio.to_thread(store.read_user, root)
     rows = await _rows(project_id, root, draft)
-    lines = linker.suggest(rows, await asyncio.to_thread(recalled, session))
-    return {"줄": lines, "저절로": len(linker.auto_pairs(lines))}
+    board = linker.board(draft, rows)
+    left = {r["key"] for r in board["남은"]}
+    memory = await asyncio.to_thread(recalled, session)
+    lines = [s for s in linker.suggest(rows, memory) if s["key"] in left]
+    return {"줄": lines, "저절로": len(linker.auto_pairs(lines)), **board}
 
 
 @router.post(PATH + "/apply")
@@ -106,7 +124,8 @@ async def apply(
 ) -> dict[str, Any]:
     from B09_Estimation.B09_Estimation_Router import payload  # 서로 부름 — 부를 때만
 
-    if body.mode not in linker.MODES:
+    modes = {p.line: p.mode for p in body.pairs if p.mode}
+    if any(m not in linker.MODES for m in (body.mode, *modes.values())):
         raise HTTPException(status_code=400, detail="수량 갈래는 take · price 중 하나")
     root = await _project_root(project_id)
     draft = _draft(body.user) or await asyncio.to_thread(store.read_user, root)
@@ -115,7 +134,7 @@ async def apply(
     if body.auto:
         memory = await asyncio.to_thread(recalled, session)
         pairs += [p for p in linker.auto_pairs(linker.suggest(rows, memory)) if p not in pairs]
-    made, report = linker.link(draft, rows, pairs, body.mode, body.counts)
+    made, report = linker.link(draft, rows, pairs, body.mode, body.counts, modes)
     made = store.clean_user(made)
     layer = await asyncio.to_thread(keep, session, report.pop("기억"))
     return {
@@ -123,3 +142,40 @@ async def apply(
         "알림": {**report, "기억층": layer},
         "payload": await payload(project_id, root, made, True),
     }
+
+
+@router.post(PATH + "/undo")
+async def undo(
+    project_id: UUID, body: UndoBody, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    from B09_Estimation.B09_Estimation_Router import payload  # 서로 부름 — 부를 때만
+
+    root = await _project_root(project_id)
+    draft = _draft(body.user) or await asyncio.to_thread(store.read_user, root)
+    made, why = linker.unlink(draft, await _rows(project_id, root, draft), body.target)
+    if why:
+        raise HTTPException(status_code=400, detail=why)
+    made = store.clean_user(made)
+    return {
+        "user": made,
+        "알림": {"풀음": 1},
+        "payload": await payload(project_id, root, made, True),
+    }
+
+
+@router.post(PATH + "/forget")
+async def forget(
+    project_id: UUID, body: ForgetBody, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    root = await _project_root(project_id)
+    draft = _draft(body.user) or await asyncio.to_thread(store.read_user, root)
+    rows = await _rows(project_id, root, draft)
+    board = linker.board(draft, rows)
+    hit = next((r for r in board["남은"] if r["key"] == body.key), None)
+    memo = linker.key_of(hit) if hit else None
+    memo = memo or next((r["memo"] for r in board["이은"] if r["key"] == body.key), None)
+    if not memo:
+        raise HTTPException(status_code=400, detail="불러온 줄 · 이은 줄이 아님")
+    layer = await asyncio.to_thread(keep, session, {memo: None})
+    left = memo in await asyncio.to_thread(recalled, session)
+    return {"잊음": memo, "기억층": layer, "남음": left}

B09_Estimation/B09_Estimation_Store.py

@@ -168,6 +168,19 @@ def _fixed_cells(row: dict[str, Any]) -> dict[str, Any]:
     return out
 
 
+def _clean_was(raw: Any) -> dict[str, Any] | None:
+    """707-4 이은 자동 줄 되돌릴 것 — 불러온 줄(hand 먼저) · 자동 줄 덮개 · 앞 · 뒤 줄 키."""
+    if not isinstance(raw, dict) or not isinstance(raw.get("rows"), list):
+        return None
+    rows = _clean_v2({"rows": raw["rows"]}).get("rows") or []
+    if not rows or not rows[0]["key"].startswith("hand:"):
+        return None
+    old = raw.get("old")
+    old = (_clean_v2({"rows": [old]}).get("rows") or [None])[0] if isinstance(old, dict) else None
+    near = {k: [x for x in raw.get(k) or [] if isinstance(x, str)][:500] for k in ("prev", "next")}
+    return {"rows": rows, "old": old, **near}
+
+
 def _clean_v2(raw: dict[str, Any]) -> dict[str, Any]:
     """184 덮개 v2 — 표 차례(겹친 키 · `fee` 버림) · 새 줄(`hand:`) = 모든 칸 ·
     자동 줄 = 고친 칸만(값 · null = 일부러 빈칸 · 못 쓰는 값 = 칸 버림) · `gone` = 자동 줄 키 ·
@@ -204,6 +217,8 @@ def _clean_v2(raw: dict[str, Any]) -> dict[str, Any]:
             if "ref" in row and (row["ref"] is None or _clean_ref(row["ref"]) is not None):
                 out["ref"] = _clean_ref(row["ref"])
             out.update(_fixed_cells(row))
+            if was := _clean_was(row.get("was")):
+                out["was"] = was
         rows.append(out)
     given = raw.get("gone") if isinstance(raw.get("gone"), list) else []
     gone = list(

resources/tester/test_707_4_bill_link.py

@@ -221,3 +221,62 @@ def test_기억_층(tmp_path, monkeypatch, role, layer):
     assert got["a||"]["확정"] == ("earth:4" if layer == "personal" else "earth:5")
     assert got["b||"]["확정"] == "x"
     assert route.recalled({"user_id": 3, "role": role}) == {}  # 회사 없음 = 층 없음
+
+
+def _same(a, b):
+    """두 셈 = 같은 줄 · 같은 직접비."""
+    assert [r["key"] for r in a["rows"]] == [r["key"] for r in b["rows"]]
+    assert a["totals"]["direct"]["sum"] == b["totals"]["direct"]["sum"]
+
+
+@pytest.mark.parametrize("mode", ["take", "price"])
+def test_잇기_풀기_제자리(geo, mode):
+    user, rows = _load(geo)
+    before = _bill(user)
+    group = next(r for r in rows if r["name"] == "콘크리트구조물")
+    pairs = [*_pick(lk.suggest(rows, {})), (group["key"], ST)]
+    made, report, bill = _link(user, rows, pairs, mode)
+    assert report["이음"] == 12 and report["묶음"] == 1
+    linked = [e for e in made["bill"]["rows"] if e.get("was")]
+    assert len(linked) == 13  # 저장 다듬기 뒤에도 되돌릴 것 남음
+    for entry in linked[::2] + linked[1::2]:  # 이은 차례와 다르게
+        made, why = lk.unlink(made, bill["rows"], entry["key"])
+        assert why == ""
+        made = store.clean_user(made)
+        bill = _bill(made)
+    _same(bill, before)
+    assert lk.unlink(made, bill["rows"], "earth:4")[1] == "이은 줄이 아님"
+
+
+def test_줄마다_갈래_판(geo):
+    user, rows = _load(geo)
+    cut = next(r for r in rows if r["name"] == "토사절취")
+    side = next(r for r in rows if r["name"] == "측구터파기" and r["spec"] == "토사")
+    pairs = [(cut["key"], "earth:4"), (side["key"], "earth:11")]
+    made, report, bill = _link(user, rows, pairs, "take", modes={side["key"]: "price"})
+    got = {r["key"]: r for r in bill["rows"]}
+    assert got["earth:4"]["qty"] == cut["qty"] and got["earth:11"]["qty"] == 100
+    with pytest.raises(ValueError):
+        lk.link(user, rows, pairs, "take", modes={cut["key"]: "both"})
+    board = lk.board(made, bill["rows"])
+    left = {r["key"] for r in board["남은"]}
+    assert cut["key"] not in left and side["key"] not in left
+    assert {r["key"] for r in board["이은"]} == {"earth:4", "earth:11"}
+    assert board["이은"][0]["memo"] in report["기억"]
+    assert not {"earth:4", "earth:11"} & {r["key"] for r in board["자리"]}
+    group = next(r for r in rows if r["name"] == "콘크리트구조물")
+    inner = {r["key"] for r in rows if r["parent"] == group["key"]}
+    made, _r, bill = _link(made, bill["rows"], [(group["key"], ST)], "take")
+    left = {r["key"] for r in lk.board(made, bill["rows"])["남은"]}
+    assert not inner & left  # 구조물 묶음에 딸린 줄 = 남은 줄 아님
+
+
+def test_이_짝_잊기(tmp_path, monkeypatch):
+    monkeypatch.setattr(route.access, "folder", lambda got, create=False: tmp_path / got.layer)
+    user = {"company_id": 7, "user_id": 3, "role": "USER"}
+    route.keep(user, {"a||": {"확정": "earth:4"}, "b||": {"확정": "earth:5"}})
+    hopyo_store.remember(tmp_path / "company", {"c||": {"확정": "x"}}, lk.MEMORY)
+    assert route.keep(user, {"a||": None}) == "personal"
+    assert set(route.recalled(user)) == {"b||", "c||"}
+    route.keep(user, {"c||": None})
+    assert "c||" in route.recalled(user)  # 일반 = 회사 기억 못 지움(남음)