From 1a4c1b5178b0561aef54f465ee6a4d7957a45130 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 31 Oct 2026 16:58:12 +0900 Subject: [PATCH] =?UTF-8?q?feat(B09=20=EC=84=A4=EA=B3=84=EB=82=B4=EC=97=AD?= =?UTF-8?q?=EC=84=9C):=20=EB=82=B4=EC=97=AD=20=EC=9E=87=EA=B8=B0=20?= =?UTF-8?q?=EC=84=9C=EB=B2=84=20=EA=B8=B8=20=EB=81=BC=EC=9B=80=20=E2=80=94?= =?UTF-8?q?=20=EC=9E=87=EA=B8=B0=20=ED=92=80=EA=B8=B0=20=C2=B7=20=EC=9D=B4?= =?UTF-8?q?=20=EC=A7=9D=20=EC=9E=8A=EA=B8=B0=20=C2=B7=20=EC=A4=84=EB=A7=88?= =?UTF-8?q?=EB=8B=A4=20=EC=88=98=EB=9F=89=20=EA=B0=88=EB=9E=98=20=C2=B7=20?= =?UTF-8?q?=EC=9E=87=EA=B8=B0=20=ED=8C=90=20(PLAN=20707-4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Router 에 Router_Link 끼움 - 이은 자동 줄 was = 불러온 줄 · price 안 줄 · 자동 줄 덮개 · 같은 부모 앞 뒤 줄 키 → 풀기 = 제자리 - link/undo · link/forget 길 · suggest 답에 남은 줄 · 빈 자리 · 이은 줄(구조물 묶음에 딸린 줄 뺌) - 시험 test_707_4 4 더함(풀기 차례 바꿔도 제자리 · 줄마다 갈래 · 판 · 일반 계정 회사 기억 못 지움) Co-Authored-By: Claude Opus 5.5 --- B09_Estimation/B09_Estimation_Bill_Link.py | 98 ++++++++++++++++++-- B09_Estimation/B09_Estimation_Router.py | 2 + B09_Estimation/B09_Estimation_Router_Link.py | 70 ++++++++++++-- B09_Estimation/B09_Estimation_Store.py | 15 +++ resources/tester/test_707_4_bill_link.py | 59 ++++++++++++ 5 files changed, 228 insertions(+), 16 deletions(-) diff --git a/B09_Estimation/B09_Estimation_Bill_Link.py b/B09_Estimation/B09_Estimation_Bill_Link.py index 5fee2e7a1..0732de6f8 100644 --- a/B09_Estimation/B09_Estimation_Bill_Link.py +++ b/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, + } diff --git a/B09_Estimation/B09_Estimation_Router.py b/B09_Estimation/B09_Estimation_Router.py index 058f9fffb..2e6424a3a 100644 --- a/B09_Estimation/B09_Estimation_Router.py +++ b/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 · 폴더째 지우면 길도 없음) — 같은 머리 · 같은 권한 diff --git a/B09_Estimation/B09_Estimation_Router_Link.py b/B09_Estimation/B09_Estimation_Router_Link.py index a5cb113f1..8ccb754c9 100644 --- a/B09_Estimation/B09_Estimation_Router_Link.py +++ b/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} diff --git a/B09_Estimation/B09_Estimation_Store.py b/B09_Estimation/B09_Estimation_Store.py index 3ba1c68b3..12026ad0b 100644 --- a/B09_Estimation/B09_Estimation_Store.py +++ b/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( diff --git a/resources/tester/test_707_4_bill_link.py b/resources/tester/test_707_4_bill_link.py index 24b03ff01..13b342a39 100644 --- a/resources/tester/test_707_4_bill_link.py +++ b/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) # 일반 = 회사 기억 못 지움(남음)