"""불러오기 재료 고름(PLAN 827-3 · 요구 811-2 · 812-1) — 서버 단독. 고름 = 사람이 그 파일 안 한 종(이름 · 규격 · 단위 · 단서 — `kind_of`)에 고른 우리 자재 · 별칭 아님 (마스터 · M01 · 별칭 칸 · 다른 파일에 안 씀 · 다른 것으로 바뀔 수 있음). 자리 = 개인 층 `실무자료/재료고름/<파일 지문>.json` {파일, 고름 {종: {키, 이름, 규격, 단위}}} — [끝내기] 뒤에도 남음 · 같은 파일(지문)을 다시 올리면 그대로 보임. 판(`census`) = 그 HP 블록마다 재료 줄 판정(`자원["재료찾기"]` · 827-2)을 종으로 묶음 — 첫 판 열 때 한 번 셈 → 산근 저장본 칸 `재료판`(HP 지울 때 같이 지움). 만들기(`using` → `M02_Sangeun_FromHopyo._emit` 이 `apply`) = 고른 종 줄: 이음 = 고른 키 · `재료찾기.고름` (문서 셈 = 우리 값 · 엑셀 값 기억 없음 · `M02_Doc_Import._attach`) · 별도계상 줄 = `고름` 만(줄 이음 · 단가 0 그대로). 문서 머리 `가져온.재료` = [{기호, 종, 원래, 판정, 방법, 키, 후보, 단서, 별도계상?, 고름?, 처음}](`entry` · 처음 = 가져온 그대로의 연결 — 고름 풀기 · 사람이 따로 바꾼 줄 가름). 이미 만든 SK(`apply_made`) = 원본 지문이 같은 SK 문서 — 가져온 그대로인 줄만(「바꿈」 · 사람이 고른 줄 · 고친 값 뺌) → 수식 길 저장(서버가 다시 셈). """ from __future__ import annotations import copy as cp import json from contextlib import contextmanager from contextvars import ContextVar from pathlib import Path from typing import Any from common_util.common_util_json import atomic_write_json from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store from M02_MasterTemplete import M02_MasterTemplete_Store as store DIRNAME = "재료고름" CENSUS = "재료판" KINDS = ( "규격 없음 — 고르기", "후보 여럿", "우리 표에 없음", "이어짐", ) # 판 차례(`M02_San_MatFind`) MAX_PICKS = 2000 _PICKS: ContextVar[dict | None] = ContextVar("재료고름", default=None) def _text(v: Any) -> str: return " ".join(str(v or "").split()) def kind_of(res: dict) -> str: """종 열쇠 — 가져온 자원 이름 · 규격 · 단위 + 판정 단서(같은 이름이라도 산근 머리 Φ800 · Φ1000 은 다른 종).""" found = res.get("재료찾기") or {} head = "|".join(_text(res.get(k)) for k in ("이름", "규격", "단위")) return head + "|" + ";".join(_text(x) for x in found.get("단서") or []) def _apart(res: dict) -> bool: from M02_MasterTemplete import M02_San_Link as link return link.apart({"이름": res.get("이름") or "", "규격": res.get("규격") or ""}) def _pick(raw: Any) -> dict | None: if not isinstance(raw, dict) or not str(raw.get("키") or "").strip(): return None return {k: _text(raw.get(k)) for k in ("키", "이름", "규격", "단위")} # ── 고름 자리(개인 층) ──────────────────────────────────────────────────── def _path(folder: str | Path, fp: str) -> Path: if not fp or not fp.isalnum() or len(fp) > 128: raise store.StoreError(400, "파일 지문") return hopyo_store._dir(folder) / DIRNAME / f"{fp}.json" def read(folder: str | Path, fp: str) -> dict[str, Any]: """{파일, 고름 {종: 고른 것}} — 없으면 빈 것.""" try: got = json.loads(_path(folder, fp).read_bytes()) except (OSError, ValueError): return {"파일": "", "고름": {}} picks = got.get("고름") if isinstance(got.get("고름"), dict) else {} return { "파일": str(got.get("파일") or ""), "고름": {k: p for k, v in picks.items() if (p := _pick(v))}, } def write(folder: str | Path, fp: str, name: str, picks: dict) -> dict[str, dict]: """종마다 고른 것(None = 고름 풀기)을 얹어 씀 → 지금 고름 전부.""" now = read(folder, fp) mine = dict(now["고름"]) for k, v in (picks or {}).items(): if (p := _pick(v)) is None: mine.pop(str(k), None) else: mine[str(k)] = p if len(mine) > MAX_PICKS: raise store.StoreError(400, "고름이 너무 많음") path = _path(folder, fp) path.parent.mkdir(parents=True, exist_ok=True) atomic_write_json(path, {"파일": name or now["파일"], "고름": mine}) return mine # ── 만들기 ─────────────────────────────────────────────────────────────── @contextmanager def using(picks: dict | None): """이 안에서 만드는 가져온 식 SK(안쪽 블록 포함)에 그 파일 고름을 얹음.""" token = _PICKS.set(picks or None) try: yield finally: _PICKS.reset(token) def apply(logic: dict) -> int: """만들 로직 줄에 고름 얹음 → 얹은 줄 수(고름 없으면 0 · 엔진 판정 그대로).""" picks, n = _PICKS.get(), 0 for item in (logic or {}).get("호표") or [] if picks else []: res = item.get("자원") or {} found = res.get("재료찾기") if not found or not (p := picks.get(kind_of(res))): continue res["재료찾기"] = {**found, "고름": p} if not _apart(res): item["이음"] = p["키"] n += 1 return n def chosen(item: dict) -> dict | None: """그 줄 고름(별도계상 아님) — 문서 셈 = 우리 값.""" res = item.get("자원") or {} got = (res.get("재료찾기") or {}).get("고름") return got if got and not _apart(res) else None def item_key(res: dict) -> str | None: """줄 품목 키 — 고름 › 엔진 이음.""" found = res.get("재료찾기") or {} return (found.get("고름") or {}).get("키") or found.get("키") def apart_note(found: dict) -> str: """별도계상 줄 설명 — 품목이 있으면 「별도계상 · 품목: 이름 규격」.""" from M02_MasterTemplete import M02_San_Link as link key = (found.get("고름") or {}).get("키") or found.get("키") if not key: return link.APART one = found.get("고름") or next( (c for c in found.get("후보") or [] if c.get("키") == key), None ) if one is None: # 엔진이 이은 품목 = 우리 자재표 줄 from M02_MasterTemplete import M02_San_MatFind as mf one = ((mf._row(key) or {}).get("row") or {}).get("one") or {} what = f"{_text(one.get('이름'))} {_text(one.get('규격'))}".strip() or key return f"별도계상 · 품목: {what} · 셈은 프로젝트에서" def entry(item: dict, sym: str, excel: Any) -> dict | None: """문서 머리 `가져온.재료` 한 칸(재료 찾기 판정 있는 줄만).""" res = item.get("자원") or {} found = res.get("재료찾기") if not found: return None one = { "기호": sym, "종": kind_of(res), "원래": {k: _text(res.get(k)) for k in ("이름", "규격", "단위")}, } one |= {k: found.get(k) for k in ("판정", "방법", "키", "후보", "단서")} if _apart(res): one["별도계상"] = True elif found.get("키"): one["처음"] = {"갈래": "요소", "키": found["키"], "엑셀": excel} else: one["처음"] = {"갈래": "숫자", "값": excel} return one | ({"고름": found["고름"]} if found.get("고름") else {}) # ── 판(그 파일 재료 종) ────────────────────────────────────────────────── def census(folder: str | Path, key: str) -> list[dict]: """HP 블록마다 재료 줄 판정 → 종 [{종, 이름, 규격, 단위, 판정, 방법, 키, 후보, 단서, 별도계상, 블록, 줄, 호표?(828-1 일위 구성 줄 · 코드 숫자)}].""" from M02_MasterTemplete import M02_Doc_FromNative as fn from M02_MasterTemplete import M02_Sangeun_FromHopyo as fh hopyo_store.read(folder, key) # 없는 HP = 404 path = fh._path(folder, key) try: saved = json.loads(path.read_bytes()) except (OSError, ValueError) as e: raise store.StoreError(409, fh.RELOAD) from e if isinstance(saved.get(CENSUS), list): return saved[CENSUS] native = saved.get("원파일") src = native if isinstance(native, dict) else fh._excel_of(path, saved) if saved.get("시트") else {} # fmt: skip groups: dict[str, dict] = {} # ponytail: 블록마다 셈 전부(기계 찾기 포함 · 70 블록 10~20 초) — 첫 판 한 번 · 느리면 재료 줄만 도는 길 for no in sorted(fn.numbers(src)) if src else []: with using(None): try: out = fn.emit(src, no) except Exception: # noqa: BLE001 — 못 옮기는 블록 = 판에서 뺌(만들기 때 그 까닭이 보임) continue for item in out["로직"].get("호표") or []: res = item.get("자원") or {} found = res.get("재료찾기") if not found: continue k = kind_of(res) g = groups.setdefault( k, {"종": k, **{x: _text(res.get(x)) for x in ("이름", "규격", "단위")}, **{x: found.get(x) for x in ("판정", "방법", "키", "후보", "단서")}, "별도계상": _apart(res), "블록": [], "줄": 0}, ) # fmt: skip g["줄"] += 1 if no not in g["블록"]: g["블록"].append(no) from M02_MasterTemplete import M02_San_Ilwi as il for b, _, res in il.materials(src) if src else (): # 828-1 일위 구성 줄(호표 = 코드 숫자) if not (found := res.get("재료찾기")): continue k = kind_of(res) g = groups.setdefault( k, {"종": k, **{x: _text(res.get(x)) for x in ("이름", "규격", "단위")}, **{x: found.get(x) for x in ("판정", "방법", "키", "후보", "단서")}, "별도계상": _apart(res), "블록": [], "줄": 0}, ) # fmt: skip g["줄"] += 1 hp = g.setdefault("호표", []) if (n := int(str(b["코드"])[1:])) not in hp: hp.append(n) out_ = sorted(groups.values(), key=lambda g: (KINDS.index(g["판정"]) if g["판정"] in KINDS else 9, -len(g["블록"]), g["이름"])) # fmt: skip atomic_write_json(path, {**saved, CENSUS: out_}) return out_ def fingerprint(folder: str | Path, key: str) -> tuple[str, str]: hp = hopyo_store.read(folder, key) return str(hp.get("지문") or ""), str(hp.get("파일") or "") # ── 이미 만든 SK ───────────────────────────────────────────────────────── def made(folder: str | Path, fp: str) -> list[str]: """그 층 가져온 식 SK 가운데 원본 지문이 같은 것.""" out = [] for path in sorted((Path(folder) / "sangeun").glob("SK-*.json")): try: doc = json.loads(path.read_bytes()) except (OSError, ValueError): continue src = ((doc.get("가져온") or {}).get("원본") or {}) if isinstance(doc, dict) else {} if fp and src.get("지문") == fp: out.append(path.stem) return out def _same(a: Any, b: Any) -> bool: """연결이 가져온 그대로(갈래 · 키 · 숫자 값).""" if not isinstance(a, dict) or not isinstance(b, dict) or a.get("바꿈"): return False if a.get("갈래") != b.get("갈래") or a.get("키") != b.get("키"): return False if a.get("갈래") == "숫자": return json.dumps(a.get("값"), default=str) == json.dumps(b.get("값"), default=str) or _num(a.get("값")) == _num(b.get("값")) # fmt: skip return a.get("갈래") != "요소" or "엑셀" in a def _num(v: Any) -> Any: try: return round(float(v), 6) except (TypeError, ValueError): return v def patch(doc: dict, picks: dict, kinds: set | None = None) -> int: """문서 `가져온.재료` 줄에 고름 얹음 · 풂 → 고친 줄 수(사람이 따로 바꾼 줄 · 「바꿈」 줄 그대로).""" import master_doc as md # `M02_Doc_Copy` 가 scripts 를 길에 둔 뒤 by = {md.var_id(v): v for v in doc.get("변수") or []} n = 0 for e in (doc.get("가져온") or {}).get("재료") or []: if kinds is not None and e.get("종") not in kinds: continue p, was = picks.get(e.get("종")), e.get("고름") var = by.get(e.get("기호")) if (p or {}).get("키") == (was or {}).get("키") or var is None: continue link = var.get("연결") or {} if e.get("별도계상"): e.pop("고름", None) if p is None else e.__setitem__("고름", p) var["설명"] = ( f"이음 없음 — {apart_note(e)} · 엑셀 단가 그대로" # `Doc_Import._attach` 꼴 ) n += 1 continue mine = was and link.get("갈래") == "요소" and link.get("키") == was.get("키") and "엑셀" not in link and not link.get("바꿈") # fmt: skip if not (mine or _same(link, e.get("처음"))): continue if p: var["연결"] = {"갈래": "요소", "키": p["키"]} e["고름"] = p else: var["연결"] = cp.deepcopy(e.get("처음")) e.pop("고름", None) n += 1 return n def apply_made(fp: str, picks: dict, kinds: set | None = None, skip: tuple = ()) -> dict[str, list]: """지금 층(`store.root`) 그 파일 SK 에 고름 얹어 저장 → {고침, 그대로}(가져온 그대로인 줄만 바뀜).""" from M02_MasterTemplete import M02_Doc_Sk_Path as skpath from M02_MasterTemplete import M02_Sangeun_Doc as sd fixed, kept = [], [] for key in made(store.root(), fp): if key in skip: continue got = store.read(sd.KIND, key) doc = cp.deepcopy(skpath.current(got["문서"])["문서"]) if not patch(doc, picks, kinds): kept.append(key) continue skpath.save(key, {"문서": doc, "판": got["판"]}) fixed.append(key) return {"고침": fixed, "그대로": kept} def uses(fp: str, kind: str, skip: tuple = ()) -> list[str]: """지금 층 그 파일 SK 가운데 그 종 줄이 있는 것(SK 상세 물음 · 저장 없음).""" from M02_MasterTemplete import M02_Sangeun_Doc as sd out = [] for key in made(store.root(), fp): if key in skip: continue doc = json.loads((store.root() / sd.KIND / f"{key}.json").read_bytes()) lines = ((doc.get("가져온") or {}).get("로직") or {}).get("호표") or [] if any(kind_of(x.get("자원") or {}) == kind for x in lines if (x.get("자원") or {}).get("재료찾기")): # fmt: skip out.append(key) return out def tally(folder: str | Path, fp: str, kind: str) -> dict | None: """그 층 작업 중 HP(같은 지문) 판에서 그 종 {블록, 줄} 수 — 판을 아직 안 셈 · HP 없음 = None.""" from M02_MasterTemplete import M02_Sangeun_FromHopyo as fh for row in hopyo_store.rows(folder): if row.get("지문") != fp: continue try: got = json.loads(fh._path(folder, row["키"]).read_bytes()).get(CENSUS) except (OSError, ValueError): continue for g in got or []: if g.get("종") == kind: return {"블록": len(g.get("블록") or []), "줄": g.get("줄") or 0} return None