Merge remote-tracking branch 'origin/dev' into sub_laptop_7

This commit is contained in:
2026-10-31 17:18:26 +09:00
16 changed files with 918 additions and 30 deletions
+1
View File
@@ -90,3 +90,4 @@ AUTH_OTP_DISABLED=True
NVIDIA_API_KEY=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
# 노선분석 산사태위험지도(행안부 생활안전지도 WMS IF_0046 · D03_PlanAnalysis_Landslide) 키 — 비우면 겹 꺼짐
SAFEMAP_API_KEY=
DATA_GO_KR_API_KEY=0j5WZHeNxSkdPVSLXUsx9kVikXEN%2BDc%2Byv68jz8IpKFffjbZkRiQrJO78y6rxPCTqhiSOW4lDxpid0twkSVQ9g%3D%3D
+2
View File
@@ -42,6 +42,8 @@ storage/
#resources/
# 전국 등고선 gpkg 22GB — 영구 추적 제외 (배포서버 구성 시 직접 배치)
resources/data_global_contours/national_contours.gpkg
# 산림 주제도 원본(산림청 공개 파일) — 추적 제외
resources/data_forest/
# 로그 파일
*.log
+344
View File
@@ -0,0 +1,344 @@
"""B09 내역 줄 ↔ 우리 공종 잇기(PLAN 707-4) — 불러온 줄(707-1 `hand:x…`)을
자동 줄(earth · st · haul)에 이어 B08 수량이 붙게. 파일 · 저장 안 만짐(맞춤 기억 읽기 · 쓰기 =
부르는 쪽이 층 폴더로).
제안(`suggest`) — 아직 안 이은 불러온 줄마다 후보 · 꼬리표:
「지난번 이음」 = 맞춤 기억 `{품명|규격|단위: {확정: 줄 키, 이름}}`(실무자료 맞춤 꼴 ·
규격이 달라도 품명 · 단위가 같고 짝이 하나면) · 키가 없는 공사 = 같은 이름 자동 줄
「짝 표」 = 토공 이름 짝 표(`PAIRS` · 코드 상수) · 우리 이름(`Bill_Items.EARTH`) 그대로도
`하나` = 한 줄 ↔ 한 줄 — 아니면(암 갈래 · 여러 줄 → 한 줄) 사용자 고름 ·
`auto_pairs` = 기억 · 하나만
잇기(`link`) — 항목 줄 ↔ 자동 항목 줄 = 707-1 붙임과 같은 꼴(불러온 줄 자리에 자동 줄 ·
고정 단가 칸 · 연결 · 수량 갈래 take = 엑셀 수량 덮개 / price = B08(없으면 엑셀) ·
불러온 줄 지움 = 두 번 셈 0)
묶음 줄 ↔ 구조물 줄(`st:`) 하나(1 : N) — take = 구조물 줄이 묶음 자리 ·
안 줄이 그 아래(저절로 묶음 · 금액 = 안 줄 합) / 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
from B09_Estimation.B09_Estimation_Bill_Import import MODES, PARTS, _flat, freeze
from B09_Estimation.B09_Estimation_Bill_Items import EARTH
from M02_MasterTemplete.M02_Hopyo_Match import line_key
MEMORY = "내역맞춤"
TABLE, RECALL = "짝 표", "지난번 이음"
HEADS = ("earth", "st", "haul")
_TAIL = re.compile(r"^(.+?)\((.*)\)$")
ROCK = ("절토(연암)", "암발파")
#: 토공 이름 짝 표 — (경쟁사 이름, 규격에 든 글 | "", 우리 이름) · 우리 이름 둘 = 한 줄 ↔ 한 줄 아님
#: (EST 암 = 연암 90 % · 보통암 10 % ↔ 우리 연암 · 경암) · 이름 괄호 글 = 규격으로도 봄
PAIRS: tuple[tuple[str, str, tuple[str, ...]], ...] = (
("수평규준틀", "", ("규준틀설치 종단규준틀",)),
("비탈규준틀", "", ("규준틀설치 횡단규준틀",)),
("뿌리뽑기", "", ("제근",)),
("토사절취", "", ("절토(토사)",)),
("암석절취", "", ROCK),
("암절취", "", ROCK),
("구조물터파기", "토사", ("구조물터파기(토사)",)),
("구조물터파기", "암", ("구조물터파기(연암)",)),
("터파기", "암", ("구조물터파기(연암)",)),
("구조물되메우기", "", ("구조물 되메우기",)),
("되메우기", "", ("구조물 되메우기",)),
("측구터파기", "토사", ("측구파기(토사)",)),
("측구터파기", "암", ("측구파기(연암)",)),
("도자운반-토사", "", ("도자운반",)),
("도자운반-암", "", ("도자운반",)),
("흙운반D/Z19Ton", "", ("도자운반",)),
("덤프운반-토사", "", ("덤프운반",)),
("덤프운반-암", "", ("덤프운반",)),
("흙운반D/P+B/H", "", ("덤프운반",)),
("토사운반", "", ("덤프운반",)),
("잔토처리", "", ("사토처리",)),
("사토운반", "", ("사토처리",)),
("사토무대처리", "", ("사토처리",)),
("노면고르기및다짐", "", ("노면정리",)),
)
EARTH_KEY = {name: f"earth:{n}" for n, (_s, name, _u, _w) in enumerate(EARTH, 1)}
def _texts(name: Any, spec: Any) -> list[tuple[str, str]]:
"""(이름, 규격) 보기 — 그대로 · 이름 끝 괄호 글을 규격 앞에."""
n, s = _flat(name), _flat(spec)
hit = _TAIL.match(n)
return [(n, s), *([(hit[1], hit[2] + s)] if hit else [])]
def table(name: Any, spec: Any) -> tuple[str, ...]:
"""짝 표 → 우리 토공 이름들(없음 = ())."""
n, s = _texts(name, spec)[0]
for ours in EARTH_KEY:
if _flat(ours) in (n, n + s):
return (ours,)
for alias, part, targets in PAIRS:
if any(t == _flat(alias) and _flat(part) in u for t, u in _texts(name, spec)):
return targets
return ()
def key_of(row: dict[str, Any]) -> str:
"""맞춤 기억 열쇠 = 품명 | 규격 | 단위(실무자료 맞춤과 같은 정규화)."""
return line_key({"이름": row.get("name"), "규격": row.get("spec"), "단위": row.get("unit")})
def _loose(key: str) -> str:
name, _spec, unit = (key.split("|") + ["", ""])[:3]
return f"{name}|{unit}"
def recall(memory: dict[str, dict[str, Any]], row: dict[str, Any]) -> dict[str, Any] | None:
"""기억 한 칸 — 같은 열쇠 · 없으면 품명 · 단위 같고 짝이 하나인 것."""
key = key_of(row)
if isinstance(memory.get(key), dict):
return memory[key]
loose = [v for k, v in memory.items() if isinstance(v, dict) and _loose(k) == _loose(key)]
return loose[0] if len({v.get("확정") for v in loose}) == 1 else None
def _targets(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""이을 수 있는 자동 줄 {키: 줄} — earth · st · haul(항목이든 저절로 묶음이든)."""
return {r["key"]: r for r in rows if r.get("auto") and r["key"].split(":")[0] in HEADS}
def _linked(rows: list[dict[str, Any]]) -> set[str]:
"""이미 이은 자동 줄 — 불러온 수량(비교 칸) 있음 · 불러온 묶음 아래 · 안 줄을 거느림."""
parents = {r.get("parent") for r in rows}
return {
k
for k, r in _targets(rows).items()
if r.get("src_qty") is not None
or str(r.get("parent") or "").startswith("hand:x")
or (k.startswith("st:") and k in parents)
}
def _resolve(got: dict[str, Any], targets: dict[str, dict[str, Any]]) -> str | None:
key = got.get("확정")
if key in targets:
return key
head = str(key or "").split(":")[0]
same = [
k for k, r in targets.items() if k.startswith(f"{head}:") and r["name"] == got.get("이름")
]
return same[0] if len(same) == 1 else None
def suggest(rows: list[dict[str, Any]], memory: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
"""[{key, kind, name, spec, unit, 후보 [{key, name}], 꼬리표, 하나, 까닭}] —
후보 있는 불러온 줄만."""
targets = _targets(rows)
free = {k: r for k, r in targets.items() if k not in _linked(rows)}
out = []
for row in rows:
if not row["key"].startswith("hand:x"):
continue
got = recall(memory, row)
hit = _resolve(got, free) if got else None
if hit:
keys, tag = [hit], RECALL
elif row["kind"] == "item":
keys, tag = [EARTH_KEY[n] for n in table(row["name"], row["spec"])], TABLE
keys = [k for k in keys if k in free]
else:
keys = []
if not keys:
continue
one = len(keys) == 1
out.append(
{
**{f: row.get(f) for f in ("key", "kind", "name", "spec", "unit")},
"후보": [{"key": k, "name": free[k]["name"]} for k in keys],
"꼬리표": tag,
"하나": one,
"까닭": "" if one else "한 줄 ↔ 한 줄 아님 — 고름",
}
)
seen: dict[str, int] = {}
for s in out:
if s["하나"]:
seen[s["후보"][0]["key"]] = seen.get(s["후보"][0]["key"], 0) + 1
for s in out: # 여러 줄 → 한 자동 줄(도자운반-토사 · -암) = 고름
if s["하나"] and seen[s["후보"][0]["key"]] > 1:
s.update({"하나": False, "까닭": "여러 줄 → 한 줄 — 고름"})
return out
def auto_pairs(suggestions: list[dict[str, Any]]) -> list[tuple[str, str]]:
"""저절로 이을 짝 [(불러온 줄 키, 자동 줄 키)] — 지난번 이음 · 하나만."""
return [
(s["key"], s["후보"][0]["key"]) for s in suggestions if s["꼬리표"] == RECALL and s["하나"]
]
def _descendants(entries: list[dict[str, Any]], key: str) -> set[str]:
out, grow = set(), {key}
while grow:
grow = {e["key"] for e in entries if e.get("parent") in grow} - out
out |= grow
return out
def _place(bill: dict[str, Any], at: int, entry: dict[str, Any], parent: Any) -> None:
entry.pop("parent", None)
if parent:
entry["parent"] = parent
bill["rows"].insert(at, entry)
if entry["key"] in bill["gone"]:
bill["gone"].remove(entry["key"])
def link(
user: dict[str, Any],
rows: list[dict[str, Any]],
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 수량) ·
`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}
targets = _targets(rows)
used = _linked(rows)
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)
why = (
"불러온 줄이 아님" if at < 0 or not line.startswith("hand:x") or row is None
else "이을 수 없는 줄" if goal is None
else "이미 이은 줄" if target in used
else "묶음 ↔ 구조물 줄만" if (row["kind"] != "item") != target.startswith("st:")
else ""
) # fmt: skip
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_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)
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:
auto["qty"] = hand.get("qty")
report["이음"] += 1
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(
{
"ref": None,
"price": {p: sum(r.get(p) or 0 for r in items) / count for p in PARTS},
"src_qty": count,
}
)
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,
}
+2
View File
@@ -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 · 폴더째 지우면 길도 없음) — 같은 머리 · 같은 권한
@@ -0,0 +1,181 @@
"""B09 내역 줄 ↔ 우리 공종 잇기 길(PLAN 707-4) — 저장 안 함(화면 캐시로 → [저장] 때 서버 정본).
POST /api/projects/{id}/estimation/link/suggest {user?}
→ {줄: 제안(`B09_Estimation_Bill_Link.suggest` · 구조물 묶음에 딸린 줄 뺌) ·
저절로: 지난번 이음 · 하나 수 · 남은 · 자리 · 이은(`board`)}
POST /api/projects/{id}/estimation/link/apply
{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_UI_Link`.
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from typing import Any
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B09_Estimation import B09_Estimation_Bill_Link as linker
from B09_Estimation import B09_Estimation_Store as store
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_Layer_Access as access
router = APIRouter()
PATH = "/{project_id}/estimation/link"
LAYERS = ("company", "personal")
def _folders(session: dict[str, Any]) -> dict[str, tuple[Path, bool]]:
"""{층: (폴더, 쓰기)} — 회사 없음 등 = 그 층 건너뜀."""
out = {}
for layer in LAYERS:
try:
got = access.grant(session, layer)
out[layer] = (access.folder(got, create=True), got.write)
except HTTPException:
continue
return out
def recalled(session: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""맞춤 기억 — 회사 위에 개인을 덮음."""
out: dict[str, dict[str, Any]] = {}
for folder, _write in _folders(session).values():
out.update(hopyo_store.memory(folder, linker.MEMORY))
return out
def keep(session: dict[str, Any], entries: dict[str, dict[str, Any]]) -> str | None:
"""이은 짝 → 회사 층(쓸 수 있으면) · 아니면 개인 층 → 쓴 층(없음 = None)."""
folders = _folders(session)
layer = next((n for n in LAYERS if n in folders and folders[n][1]), None)
if entries and layer:
hopyo_store.remember(folders[layer][0], entries, linker.MEMORY)
return layer if entries else None
def _draft(raw: Any) -> dict[str, Any] | None:
return store.clean_user(raw) if isinstance(raw, dict) else None
class SuggestBody(BaseModel):
user: dict[str, Any] | None = None
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):
pairs: list[Pair] = Field(default_factory=list, max_length=2000)
auto: bool = False
mode: str = "take"
counts: dict[str, float] | None = None
user: dict[str, Any] | None = None
async def _rows(project_id: UUID, root: Path, user: dict[str, Any]) -> list[dict[str, Any]]:
from B09_Estimation.B09_Estimation_Router import payload # 서로 부름 — 부를 때만
return (await payload(project_id, root, user, True))["sheets"]["bill"]["rows"]
@router.post(PATH + "/suggest")
async def suggest(
project_id: UUID, body: SuggestBody, 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)
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")
async def apply(
project_id: UUID, body: ApplyBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
from B09_Estimation.B09_Estimation_Router import payload # 서로 부름 — 부를 때만
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)
rows = await _rows(project_id, root, draft)
pairs = [(p.line, p.target) for p in body.pairs]
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, modes)
made = store.clean_user(made)
layer = await asyncio.to_thread(keep, session, report.pop("기억"))
return {
"user": made,
"알림": {**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}
+15
View File
@@ -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(
@@ -254,8 +254,9 @@ async function post(path: string, body: Row): Promise<Row> {
const COST_PART = { 노무비: "labor", 재료비: "material", 경비: "expense" } as const;
/** 붙인 줄 다시 잇기(705-2) — 후보 = 서버(M02 이음 짝) · 단가 = 이 공사 단가 책 · 바꿈 = 캐시 + 서버 셈 */
function relinkFold(ctx: TabContext, cover: string): RelinkSource {
const added = (): Row[] => ((ctx.user.daega_cover || {})[cover] || {})[ADD] || [];
function relinkFold(ctx: TabContext, cover: string, from: Row): RelinkSource {
let now = from; // 붙인 뒤 캐시(ctx.user 는 그리기 전 값) — 바꿀 때마다 새 값
const added = (): Row[] => ((now.daega_cover || {})[cover] || {})[ADD] || [];
const book = (kind: "노임" | "자재", key: string): number | null | undefined =>
(ctx.payload.inputs?.[kind === "노임" ? "labor" : "materials"] || []).find(
(r: Row) => r.key === key,
@@ -302,9 +303,10 @@ function relinkFold(ctx: TabContext, cover: string): RelinkSource {
if ("노임" in c) ln.link.labor = c.노임 ?? null;
if ("자재" in c) ln.link.material = c.자재 ?? null;
}
const all: Row = JSON.parse(JSON.stringify(ctx.user.daega_cover || {}));
const all: Row = JSON.parse(JSON.stringify(now.daega_cover || {}));
all[cover] = { ...(all[cover] || {}), [ADD]: lines };
const user = { ...ctx.user, daega_cover: all };
const user = { ...now, daega_cover: all };
now = user;
Object.assign(ctx.payload, await post("calc", { user })); // 새 이음 키 = 이 공사 단가 책 수요
ctx.setUser(user);
return { back, note: `바뀐 줄 ${back.length} — [저장]으로 남김` };
@@ -333,7 +335,7 @@ async function pasteFold(ctx: TabContext, cover: string, text: string): Promise<
`고정 줄 ${got.줄} 붙임 · 엑셀 합계 ${won} · 이음 ${got.이음} · 후보 ${got.이음후보} — [저장]으로 남김 · 이 알림을 누르면 이 공사 단가로 다시 잇기`,
"success",
9000,
() => void openRelink(relinkFold(ctx, cover)),
() => void openRelink(relinkFold(ctx, cover, got.user)),
);
} catch (error) {
showToast(`붙여넣기 실패 — ${error instanceof Error ? error.message : String(error)}`, "error");
+22 -4
View File
@@ -113,6 +113,23 @@ def check_cols(cols: Any) -> dict[str, int]:
return out
def fit(cols: dict[str, int], rows: list[list[str]]) -> dict[str, int]:
"""기억한 열 짝이 다른 꼴에 붙을 때 — 칸 전체를 -1 · 0 · +1 밀어 보고 품명 글 · 수량 수가 가장 많이
맞는 쪽(엑셀꼴 「호표」 칸 있음 · STmate 꼴 없음 · 같으면 그대로)."""
def score(d: int) -> int:
got = 0
for r in rows:
name, qty = _cell(r, cols["품명"] + d), _cell(r, cols["수량"] + d)
got += bool(name and hx._num(name) is None) + (hx._num(qty) is not None)
return got
best = max((0, 1, -1), key=lambda d: (score(d), d == 0))
if best == 0 or min(cols.values()) + best < 0:
return cols
return {k: v + best for k, v in cols.items()}
def _money_cols(cols: dict[str, int]) -> dict[str, int]:
out = {"노무비": cols.get("노"), "재료비": cols.get("재"), "경비": cols.get("경")}
if cols.get("금액") is not None:
@@ -148,7 +165,7 @@ def read_text(text: str, cols: Any = None) -> dict[str, Any]:
below = [_flat(v) for v in rows[start]] if start < len(rows) else []
start += "단가" in below or "금액" in below
elif cols is not None:
used = check_cols(cols)
used = fit(check_cols(cols), rows)
else:
used, asked = guess(rows), True
money = _money_cols(used)
@@ -206,9 +223,10 @@ def read_text(text: str, cols: Any = None) -> dict[str, Any]:
qty = hx._num(_cell(row, used["수량"]))
if not unit and not any(v for v in (*price.values(), *amount.values())):
continue # 근거 줄 「건설표준품셈 | 6-7-2 | 0」
if not hopyos: # 첫 줄 수량 없음 = 호표 머리(「제N호표」 칸 없이 복사)
open_hopyo(0, row if qty is None else None)
if qty is None:
if not hopyos: # 첫 줄 수량 없음 + 금액 있음 = 엑셀꼴 호표 머리(「제N호표」 칸 없이 복사)
head_row = qty is None and any(amount.values())
open_hopyo(0, row if head_row else None)
if head_row:
continue
remark = _cell(row, note)
ref = hx._REF.match(_flat(remark))
+9 -6
View File
@@ -120,21 +120,24 @@ def delete(folder: str | Path, key: str) -> None:
# ── 맞춤 기억 ─────────────────────────────────────────────────────────
def memory(folder: str | Path) -> dict[str, dict[str, Any]]:
"""층 맞춤 기억 `{줄 열쇠: 확정}` — 없으면 빈 것."""
doc = _load(_dir(folder) / f"{MEMORY}.json")
def memory(folder: str | Path, name: str = MEMORY) -> dict[str, dict[str, Any]]:
"""층 맞춤 기억 `{줄 열쇠: 확정}` — 없으면 빈 것 ·
`name` = 기억 파일(내역 잇기 707-4 = 「내역맞춤」)."""
doc = _load(_dir(folder) / f"{name}.json")
got = doc.get("줄") if isinstance(doc, dict) else None
return got if isinstance(got, dict) else {}
def remember(folder: str | Path, entries: dict[str, dict[str, Any] | None]) -> dict[str, Any]:
def remember(
folder: str | Path, entries: dict[str, dict[str, Any] | None], name: str = MEMORY
) -> dict[str, Any]:
"""기억 더함 · 바꿈(값 None = 지움) → 새 기억 전부."""
with _LOCK:
got = memory(folder)
got = memory(folder, name)
for key, value in entries.items():
if value is None:
got.pop(key, None)
else:
got[key] = value
atomic_write_json(_dir(folder) / f"{MEMORY}.json", {"줄": got})
atomic_write_json(_dir(folder) / f"{name}.json", {"줄": got})
return got
@@ -97,7 +97,8 @@ export async function openRelink(src: RelinkSource): Promise<void> {
const load = async (): Promise<void> => {
rows = await src.rows();
pick = rows.map((r) => (r.후보.length && !linked(r) ? 0 : r.후보.length));
on = rows.map((r) => !!r.후보.length && !linked(r));
// 저절로 고름 = 후보 하나 · 붙인 단가 있음(별도계상 0 원 · 후보 여럿은 사람이 고름)
on = rows.map((r) => r.후보.length === 1 && !linked(r) && r.붙인[COST[r.후보[0].비목]] > 0);
};
const choice = (i: number): RelinkRow["후보"][number] | "keep" | "free" => {
const r = rows[i];
@@ -146,19 +147,19 @@ export async function openRelink(src: RelinkSource): Promise<void> {
});
const table = createDataTable({
columns: [
{ key: "name", label: "품명 · 규격", width: 220 },
{ key: "unit", label: "단위", width: 48 },
{ key: "qty", label: "수량", kind: "number", decimals: 4, trim: true, width: 70 },
{ key: "pasted", label: "붙인 단가", kind: "number", decimals: 1, width: 90 },
{ key: "now", label: src.priceLabel, width: 100 },
{ key: "cand", label: "이음 후보 (누름 = 다음)", width: 260 },
{ key: "price", label: "후보 단가", kind: "number", decimals: 1, width: 90 },
{ key: "gap", label: "금액 차이", kind: "number", decimals: 1, width: 90 },
{ key: "pick", label: "고름", width: 48 },
{ key: "name", label: "품명 · 규격", width: 190 },
{ key: "pick", label: "고름", width: 44 },
{ key: "unit", label: "단위", width: 44 },
{ key: "qty", label: "수량", kind: "number", decimals: 4, trim: true, width: 60 },
{ key: "pasted", label: "붙인 단가", kind: "number", decimals: 1, width: 84 },
{ key: "now", label: src.priceLabel, width: 84 },
{ key: "cand", label: "이음 후보 (누름 = 다음)", width: 210 },
{ key: "price", label: "후보 단가", kind: "number", decimals: 1, width: 84 },
{ key: "gap", label: "금액 차이", kind: "number", decimals: 1, width: 84 },
],
rows: data,
frozen: 1,
empty: "고정 줄 없음",
empty: "",
onCellClick: (i, key) => {
const r = rows[i];
if (key === "cand") {
@@ -83,7 +83,7 @@ export async function toItem(o: ToItemOptions): Promise<void> {
close();
document.dispatchEvent(new Event(ITEM_BUILT));
showToast(
`구조물 조합 「${got.이름}」 만듦 — 「실무자료」 폴더 · 줄 ${got.줄수}` +
`구조물 조합 「${label}」 만듦 — 「실무자료」 폴더 · 줄 ${got.줄수}` +
(got.열수 > 1 ? ` · 규격 열 ${got.열수}` : ""),
"success",
6000,
@@ -1025,3 +1025,7 @@ CPU는 더늘리기는 힘들어보이고 GPU 사용을 위해 부품사용에
## 요구 514 (10-31 · 사용자)
해상도가 높은건 나중에 공공기관과 협업을 하면서 정식으로 요청을 해야할것같아. 일단은 되는 범위에서 진행하는게 좋을것같은데 지금 우리프로그램에서 상당히 중요하게 보고 있는 요소가 속도야. 그래서 등고선이나 미리 계산해야하는것들을 미리 받아 계산하고 있는데 현재 위 데이터를 받으면 용량이 어느정도 되는거야? / 지형 · 0차곡 이웃 목록 밖이라는게 무슨 말이야? / 현재 위데이터를 순차적으로 받으면 랩탑에 무리가 갈까? / 리소스 폴더는 시놀로지 문제 없음. PC 스토리지 문제 없음.. 공공데이터포털의 API키를 env파일에 넣어둔걸로 기억함. 확인
## 요구 515 (10-31 · 사용자)
1. api키는 env파일로 옮겨둬. 2. 공공데이터 포털에서 데이터를 다운받는데 위치는 리소스 폴더가 맞을것같아. 적당히 폴더를 만들어서 넣어줘. 3. 지금 랩탑의 CPU 사용율이 89%로 웹앱에는 표기되는데 작업관리자로 확인해보니까 이용률이 19%밖에 안됨.... 왜 그런지 이유를 찾아봐.
+1 -1
View File
@@ -21,7 +21,7 @@
## 공공데이터포털 (data.go.kr) — 일반 인증키
- 용도: 조달청 나라장터 가격정보현황서비스 등 data.go.kr OpenAPI 전반 (API별 활용신청 필요)
- 인증키(Encoding형, URL에 그대로 사용): `0j5WZHeNxSkdPVSLXUsx9kVikXEN%2BDc%2Byv68jz8IpKFffjbZkRiQrJO78y6rxPCTqhiSOW4lDxpid0twkSVQ9g%3D%3D`
- 인증키(Encoding형) → .env DATA_GO_KR_API_KEY 로 옮김
- 발급: 2026-08-14, 계정 = data.go.kr 회원(사용자 개인). 만료 시 마이페이지 → 오픈API → 개발계정에서 갱신
- 검증(2026-08-14): 가격정보현황서비스 실호출 정상 (resultCode 00)
- 엔드포인트: `http://apis.data.go.kr/1230000/ao/PriceInfoService/<오퍼레이션>?serviceKey=<키>&pageNo=1&numOfRows=999&type=json`
@@ -92,7 +92,13 @@ def read_key(pattern, name):
def read_g2b_key():
return read_key(r"인증키\(Encoding형[^`]*`([^`]+)`", "공공데이터포털")
from dotenv import load_dotenv
load_dotenv(Path(__file__).resolve().parents[4] / ".env")
key = os.environ.get("DATA_GO_KR_API_KEY", "").strip()
if not key:
sys.exit("공공데이터포털 키 없음 — .env DATA_GO_KR_API_KEY")
return key
def collect_docs():
+28 -1
View File
@@ -334,7 +334,9 @@ def test_구조물_조합으로_규격_묶음_울진_돌기슭막이(client):
if h["번"] in (17, 24, 25)
}
for no, c in ((17, "1.5"), (24, "2.0"), (25, "2.5")):
assert abs(_total(doc, c) - excel[no]) <= excel[no] * 1e-4 # 열마다 = 그 호표 엑셀 합계(끝수 차)
assert (
abs(_total(doc, c) - excel[no]) <= excel[no] * 1e-4
) # 열마다 = 그 호표 엑셀 합계(끝수 차)
one = client.post("/api/m02/layers/personal/hopyo-item", json={**body, "묶음": []}).json()
single, _ = _item_doc(one["열"])
assert not single.get("규격") and one["이름"].endswith("H=1.5 L3=45cm")
@@ -357,3 +359,28 @@ def test_규격_열_이름():
assert item.columns(["H=1.5 L3=45cm", "H=2.0 L3=45cm"]) == ("H", ["1.5", "2.0"])
assert item.columns(["일반형", "ㄷ형"]) == ("규격", ["일반형", "ㄷ형"])
assert item.columns(["D800", "D1000"]) == ("D", ["800", "1000"])
def test_줄만_첫_줄_수량_없음_별도계산은_줄():
rows = _block(_rows("거창"), 2)[
1:
] # 「제2호표」 머리 줄 빼고 — 첫 줄 레미콘(별도계산 · 수량 없음)
got = hp.read_text(
_text(rows),
{"품명": 1, "규격": 2, "수량": 3, "단위": 4, "금액": 6, "노": 7, "재": 9, "경": 11},
)
(one,) = got["호표"]
assert one["번"] == 0 and one["이름"] == "붙여넣기" and one["줄"][0]["이름"] == "레미콘"
assert len(one["줄"]) == 5
def test_기억한_열_짝_한_칸_밀림_맞춤():
stm = {"품명": 0, "규격": 1, "수량": 2, "단위": 3, "금액": 5, "노": 6, "재": 8, "경": 10}
rows = _block(_rows("거창"), 2)[1:] # 엑셀꼴(앞 호표 칸) 줄만 — STmate 꼴로 기억한 짝
got = hp.read_text(_text(rows), stm)
assert got["열"]["품명"] == 1 and [x["이름"] for x in got["호표"][0]["줄"]][:2] == [
"레미콘",
"콘크리트공",
]
same = hp.read_text(_text(_block(_rows("대흥"), 17)), stm)
assert same["열"] == stm
+282
View File
@@ -0,0 +1,282 @@
"""707-4 내역 줄 ↔ 우리 공종 잇기 — 짝 표 · 맞춤 기억 · 묶음 통째 잇기(`Bill_Link`) ·
기억 층(`Router_Link`).
- 짝 표 = 우리 토공 줄(`Bill_Items.EARTH`) 모두 · 실무 이름(EST 거창 · STmate 봉화) → 우리 줄 ·
암 갈래(EST 연암 · 보통암) = 후보 둘 · 하나 아님 · 여러 줄 → 한 줄 = 하나 아님
- 맞춤 기억 = 거창 엑셀에서 이은 짝 → 같은 회사 장수 · 진안 .estx 저절로 이음 · 두 번 셈 0
- 수량 갈래 take = 엑셀 수량 덮개 / price = B08 · 엑셀 = 비교 칸
- 묶음 ↔ 구조물 줄 — take = 안 줄이 구조물 줄 아래 / price = 1 단위 고정 단가 · 안 줄 지움 ·
금액 = 엑셀
- 못 잇는 짝(남의 줄 · 자재 줄 · 이미 이음 · 꼴 다름) · 기억 층 = 회사 관리자 회사 · 아니면 개인
- 셈 짝(`Calc_Bill` py · ts) 안 바뀜 — 거울은 test_707_1 · test_145_3 그대로
"""
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from B09_Estimation import B09_Estimation_Bill_Import as imp # noqa: E402
from B09_Estimation import B09_Estimation_Bill_Link as lk # noqa: E402
from B09_Estimation import B09_Estimation_Estx as ex # noqa: E402
from B09_Estimation import B09_Estimation_Router_Link as route # noqa: E402
from B09_Estimation import B09_Estimation_Store as store # noqa: E402
from B09_Estimation.B09_Estimation_Bill_Excel import read_bill # noqa: E402
from B09_Estimation.B09_Estimation_Bill_Items import EARTH, earth_items # noqa: E402
from B09_Estimation.B09_Estimation_Calc_Bill import compute_estimation # noqa: E402
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store # noqa: E402
K = ROOT / "resources" / "knowledge" / "original" / "실무문서"
GEO = (
K / "01.산불거창(거창북상병곡)/01.내역서/00.25년_산불_거창(거창,북상,병곡,산8외)내역+수량.xlsx"
)
BONG = K / "04.산불 봉화/01. 내역서/2024년 산불진화임도 실시설계(기번41_현동)_ L=3.46km.xlsx"
ESTX = {
"장수": K / "02.산불 장수/01.내역서/00.단가변경_26년_산불_무주(장수,장수,덕산,산54-1).estx",
"진안": K / "03.산불진안/01.내역서/00.단가변경_26년_산불_무주(진안,백운,백암,산1-2).estx",
}
ST = "st:pipe:box|1"
ITEMS = [{**i, "qty": 100} for i in earth_items(None)] + [
{"key": ST, "section": "2.2", "name": "BOX 암거", "unit": "m", "qty": 10,
"ref": {"tab": "sg_list", "key": "X"}},
] # fmt: skip
INPUTS = {"materials": [], "bill": {"items": ITEMS}, "hopyo": {"X": {
"labor": 100, "material": 0, "expense": 0}}} # fmt: skip
def _bill(user):
return compute_estimation(INPUTS, user, None, {})["bill"]
def _load(parsed):
"""707-1 먼저 싣기(붙임 없음) → (사용자 값, 셈 줄)."""
user, _ = imp.apply(parsed, {}, _bill({})["rows"], "abcd1234", attach=False)
user = store.clean_user(user)
return user, _bill(user)["rows"]
def _link(user, rows, pairs, mode="take", **kw):
made, report = lk.link(user, rows, pairs, mode, **kw)
made = store.clean_user(made)
return made, report, _bill(made)
@pytest.fixture(scope="module")
def geo():
return read_bill(GEO.name, GEO.read_bytes())
def _pick(suggestions):
"""사용자 고름 꼴 — 보통암 = 둘째(암발파) · 그 밖 = 첫 후보 · 자동 줄 하나에 한 줄."""
out, used = [], set()
for s in suggestions:
keys = [c["key"] for c in s["후보"]]
goal = keys[1] if len(keys) > 1 and "보통암" in s["spec"] else keys[0]
if goal not in used:
out.append((s["key"], goal))
used.add(goal)
return out
def test_짝_표_우리_토공_줄_모두():
names = {name for _s, name, _u, _w in EARTH}
assert len(names) == len(EARTH) == 17
assert all(lk.table(n, "") == (n,) for n in names) # 우리 이름 그대로
assert all(set(t) <= names for _a, _p, t in lk.PAIRS) # 짝 표 = 있는 줄만
assert set(lk.EARTH_KEY.values()) == {f"earth:{n}" for n in range(1, 18)}
def test_짝_표_실무_이름_암_갈래(geo):
user, rows = _load(geo)
got = {(s["name"], s["spec"]): s for s in lk.suggest(rows, {})}
one = {k: s["후보"][0]["name"] for k, s in got.items() if s["하나"]}
assert one == {
("토사절취", "백호0.7㎥"): "절토(토사)",
("측구터파기", "토사"): "측구파기(토사)",
("성토사면다짐", "B/H0.7+유압식콤팩터"): "성토사면다짐",
("토사운반", "굴삭기+덤프+굴삭기"): "덤프운반",
("수평규준틀", ""): "규준틀설치 종단규준틀",
("비탈규준틀", ""): "규준틀설치 횡단규준틀",
("구조물 터파기(토사)", "육샹토사(0~2m,굴삭기0.2)"): "구조물터파기(토사)",
("터파기(암)", ""): "구조물터파기(연암)",
("되메우기", "백호0.7㎥(100%)"): "구조물 되메우기",
}
rock = got[("암석절취", "보통암")]
assert [c["name"] for c in rock["후보"]] == ["절토(연암)", "암발파"]
assert not rock["하나"] and rock["까닭"] == "한 줄 ↔ 한 줄 아님 — 고름"
side = got[("측구터파기", "연암")] # 연암 · 보통암 둘 → 측구파기(연암) 하나
assert not side["하나"] and side["까닭"] == "여러 줄 → 한 줄 — 고름"
assert all(s["꼬리표"] == lk.TABLE for s in got.values())
bong = lk.suggest(_load(read_bill(BONG.name, BONG.read_bytes()))[1], {})
named = {s["name"]: s["후보"][0]["name"] for s in bong if s["하나"]}
assert named["토사절취(흙깎기)"] == "절토(토사)"
assert named["구조물터파기(육상암0~1m)"] == "구조물터파기(연암)"
assert named["비탈규준틀(@60M)"] == "규준틀설치 횡단규준틀"
def test_맞춤_기억_두_번째_공사_저절로(geo, tmp_path):
user, rows = _load(geo)
made, report, bill = _link(user, rows, _pick(lk.suggest(rows, {})))
assert report["이음"] == 12 and report["못함"] == []
hopyo_store.remember(tmp_path, report["기억"], lk.MEMORY)
assert hopyo_store.memory(tmp_path) == {} # 실무자료 맞춤과 다른 파일
memory = hopyo_store.memory(tmp_path, lk.MEMORY)
assert memory["암석절취|보통암|m3"]["확정"] == "earth:6"
for name, path in ESTX.items():
parsed = ex.read_bill(path.name, path.read_bytes())
user, rows = _load(parsed)
lines = lk.suggest(rows, memory)
assert sum(s["꼬리표"] == lk.RECALL for s in lines) == 12, name
pairs = lk.auto_pairs(lines)
assert len(pairs) == 11, name # 측구 연암 = 보통암 줄과 한 자동 줄 → 고름
made, report, bill = _link(user, rows, pairs)
assert report["이음"] == 11 and report["못함"] == []
keys = [r["key"] for r in bill["rows"]]
assert len(keys) == len(set(keys)) # 두 번 셈 0 — 줄 하나씩
assert not {k for k, _t in pairs} & set(keys) # 이은 불러온 줄 = 지움
# take = 엑셀 수량 · 고정 단가 → 직접비(아는 금액) = 파일 합 + 안 이은 구조물 줄
assert bill["totals"]["direct"]["sum"] == parsed["합계"]["직접"] + 1000, name
assert lk.auto_pairs(lk.suggest(bill["rows"], memory)) == [] # 다시 = 이음 없음
def test_수량_갈래_가_나(geo):
user, rows = _load(geo)
line = next(r for r in rows if r["name"] == "토사절취")
for mode, qty in (("take", 7641), ("price", 100)):
made, report, bill = _link(user, rows, [(line["key"], "earth:4")], mode)
cut = next(r for r in bill["rows"] if r["key"] == "earth:4")
assert (cut["qty"], cut["src_qty"], cut["fixed"]) == (qty, 7641, True)
assert cut["parent"] == line["parent"] # 불러온 자리
assert cut["price_sum"] == line["price_sum"]
assert line["key"] not in {r["key"] for r in bill["rows"]}
def test_묶음_통째_잇기(geo):
user, rows = _load(geo)
base = _bill(user)["totals"]["direct"]["sum"]
group = next(r for r in rows if r["name"] == "콘크리트구조물")
inner = [r["key"] for r in rows if r["parent"] == group["key"]]
made, report, bill = _link(user, rows, [(group["key"], ST)], "take")
got = {r["key"]: r for r in bill["rows"]}
assert report["묶음"] == 1 and group["key"] not in got
assert got[ST]["kind"] == "group" and got[ST]["sum"] == group["sum"]
assert all(got[k]["parent"] == ST for k in inner) # 안 줄 = 딸려 감(1 : N)
assert bill["totals"]["direct"]["sum"] == base - 1000 # 구조물 줄 제 금액 = 안 줄 합으로
assert report["기억"] == {"콘크리트구조물||": {"확정": ST, "이름": "BOX 암거"}}
made, report, bill = _link(user, rows, [(group["key"], ST)], "price")
got = {r["key"]: r for r in bill["rows"]}
assert report["지움"] == len(inner) and not set(inner) & set(got)
assert got[ST]["kind"] == "item" and (got[ST]["qty"], got[ST]["src_qty"]) == (10, 10)
assert abs(got[ST]["sum"] - group["sum"]) <= 3 # B08 수량 × 1 단위 단가 · 끝수
made, report, bill = _link(user, rows, [(group["key"], ST)], "price", counts={group["key"]: 5})
assert next(r for r in bill["rows"] if r["key"] == ST)["sum"] == pytest.approx(
group["sum"] * 2, abs=3
) # 경쟁사 5 개 · B08 10 개 = 두 배
memory = {"콘크리트구조물||": {"확정": "st:other", "이름": "BOX 암거"}}
assert lk.auto_pairs(lk.suggest(rows, memory)) == [(group["key"], ST)] # 키 없음 = 같은 이름
def test_못_잇는_짝(geo):
user, rows = _load(geo)
line = next(r for r in rows if r["name"] == "토사절취")
group = next(r for r in rows if r["name"] == "콘크리트구조물")
pairs = [
("earth:3", "earth:4"),
(line["key"], "mat:x"),
(line["key"], ST),
(group["key"], "earth:4"),
(line["key"], "earth:4"),
(next(r for r in rows if r["name"] == "수평규준틀")["key"], "earth:4"),
]
_made, report, _bill = _link(user, rows, pairs)
assert [m["까닭"] for m in report["못함"]] == [
"불러온 줄이 아님",
"이을 수 없는 줄",
"묶음 ↔ 구조물 줄만",
"묶음 ↔ 구조물 줄만",
"이미 이은 줄",
]
assert report["이음"] == 1
with pytest.raises(ValueError):
lk.link(user, rows, [], "both")
assert lk.link(user, rows, [], "take")[1]["이음"] == 0
@pytest.mark.parametrize(
"role, layer", [("USER", "personal"), ("ADMIN", "company")], ids=["일반", "회사관리자"]
)
def test_기억_층(tmp_path, monkeypatch, role, layer):
monkeypatch.setattr(route.access, "folder", lambda got, create=False: tmp_path / got.layer)
session = {"company_id": 7, "user_id": 3, "role": role}
assert route.keep(session, {"a||": {"확정": "earth:4", "이름": "절토(토사)"}}) == layer
assert route.keep(session, {}) is None
hopyo_store.remember(
tmp_path / "company", {"a||": {"확정": "earth:5"}, "b||": {"확정": "x"}}, lk.MEMORY
)
got = route.recalled(session)
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) # 일반 = 회사 기억 못 지움(남음)