diff --git a/M02_MasterTemplete/M02_Doc_Sk_Path.py b/M02_MasterTemplete/M02_Doc_Sk_Path.py index 6ff696a0f..a7cd41b01 100644 --- a/M02_MasterTemplete/M02_Doc_Sk_Path.py +++ b/M02_MasterTemplete/M02_Doc_Sk_Path.py @@ -40,11 +40,10 @@ def _file(key: str) -> dict[str, Any]: def _settled(doc: dict, job: Any = None) -> tuple[dict, Any]: """옛 셈 길 첫머리(`calc_doc`)와 같게 — (정리한 문서, 사본 색인).""" - from M01_MasterData import M01_MasterData_Store as m01 from M02_MasterTemplete import M02_Sangeun_Calc as calc raw = {**doc, skc.JOB: job} if job else doc - whole = skc.over(doc := skc.settle(raw), m01.loaded()[1]) + whole = skc.over(doc := skc.settle(raw), skc.whole_()) # 내 기계(821-1) try: return sd.clean(doc, None, whole), whole except sd.DocError as e: diff --git a/M02_MasterTemplete/M02_Item_Prices.py b/M02_MasterTemplete/M02_Item_Prices.py index 859b53c3d..287b0e2fc 100644 --- a/M02_MasterTemplete/M02_Item_Prices.py +++ b/M02_MasterTemplete/M02_Item_Prices.py @@ -23,6 +23,7 @@ from decimal import Decimal from typing import Any from M01_MasterData import M01_MasterData_Store as m01 +from M02_MasterTemplete import M02_MyMachine as mym mf = m01.cm.mf mk = m01.mk @@ -239,6 +240,8 @@ def collect( `extra` = 마스터 밖 로직 줄(굳힌 단가산출 719-1) — 그 줄이 닿는 것만 모음(줄은 안 담음) · `over` = M02 로직 사본 줄(795-4) — 같은 키는 M01 줄 대신 · M01 에 없는 키 = `COPY_FILE`.""" files, whole = master() + whole = mym.whole(whole) # 지금 층 내 기계 줄(821-1) — 없으면 M01 그대로 + own = {str(r.get("키")): r for r in mym.added(whole)} mine = {str(r.get("키")): r for r in over or []} if mine: from M02_MasterTemplete import M02_Sangeun_Copy as skc @@ -263,7 +266,8 @@ def collect( seen.update(str(row.get("키")) for row in rows) if rows: out[name] = {**{k: v for k, v in data.items() if k != slot}, slot: rows} - if left := [r for k, r in mine.items() if k in collector.keys and k not in seen]: + left = [r for k, r in {**own, **mine}.items() if k in collector.keys and k not in seen] + if left: out[COPY_FILE] = {"그룹": "사본", "줄": left} if collector.missing: out["_빠진"] = sorted(set(collector.missing)) diff --git a/M02_MasterTemplete/M02_Machine_Find.py b/M02_MasterTemplete/M02_Machine_Find.py index 23e934bbe..b0de98932 100644 --- a/M02_MasterTemplete/M02_Machine_Find.py +++ b/M02_MasterTemplete/M02_Machine_Find.py @@ -1,6 +1,6 @@ """M02 기계 찾기(PLAN 812-9 · 요구 718 · 719) — 가져온 기계 글(이름 · 규격) → 우리 EQ · 서버 단독. -과녁 = M01 기계표 전체(EQ) · 부르는 곳 = 불러오기 이음(`M02_Hopyo_Row.m01_picks` · `M02_San_Emit._m01`) · +과녁 = M01 기계표 전체(EQ) + 지금 층 내 기계(821-1 · `M02_MyMachine.whole`) · 부르는 곳 = 불러오기 이음(`M02_Hopyo_Row.m01_picks` · `M02_San_Emit._m01`) · 일위대가 호표(`M02_Hopyo_Elem.machine(s)` · `M02_Hopyo_Fixed._eqs`). 차례: 1. 별칭(M01 기계 줄 「별칭」 칸 · 815-1 `M02_Elem_Alias`) — 「이름 규격」 통째 별칭 = 그 EQ(여럿 = @@ -22,8 +22,6 @@ import unicodedata from functools import lru_cache from typing import Any -from M01_MasterData import M01_MasterData_Store as m01 - # 소리만 다른 표기 → 한 글(양쪽에 같이 씀 · 차례대로) SOUND = ( ("로우더", "로더"), ("로울러", "롤러"), ("브레이카", "브레이커"), ("브레카", "브레이커"), @@ -142,20 +140,23 @@ class Row: ) # 규격 글 낱말(「더블」 · 「엔진식」) -@lru_cache(maxsize=2) -def _table(_whole: int) -> tuple[list[Row], frozenset[str], dict[str, Row]]: - rows = [Row(r) for r in m01.loaded()[1].numbers.get("EQ", {}).values()] +@lru_cache(maxsize=4) +def _table(whole: Any) -> tuple[list[Row], frozenset[str], dict[str, Row]]: + rows = [Row(r) for r in whole.numbers.get("EQ", {}).values()] words = frozenset(w for r in rows for w in r.words) return rows, words, {r.key: r for r in rows} -_SEEN: list = [0.0, None] # (때, 표) — M01 판 확인(파일 도장)은 1 초에 한 번 +_SEEN: list = [0.0, None, None] # (때, 표, 층) — M01 · 내 기계 판 확인은 층마다 1 초에 한 번 def table() -> tuple[list[Row], frozenset[str], dict[str, Row]]: - now = time.monotonic() - if _SEEN[1] is None or now - _SEEN[0] > 1.0: - _SEEN[:] = [now, _table(id(m01.loaded()[1]))] + from M02_MasterTemplete import M02_MasterTemplete_Store as store + from M02_MasterTemplete import M02_MyMachine as mym + + now, here = time.monotonic(), store.root() + if _SEEN[1] is None or now - _SEEN[0] > 1.0 or _SEEN[2] != here: + _SEEN[:] = [now, _table(mym.whole()), here] return _SEEN[1] diff --git a/M02_MasterTemplete/M02_Machine_Hourly.py b/M02_MasterTemplete/M02_Machine_Hourly.py index f9795dd0a..03aa70c0b 100644 --- a/M02_MasterTemplete/M02_Machine_Hourly.py +++ b/M02_MasterTemplete/M02_Machine_Hourly.py @@ -5,6 +5,8 @@ (`로직(GC, 기계='원문번호')` · 옵션은 기본값). 짝 차례 · 셈은 `DataFlow/M02_DF_Values._machine_hourly` 와 같은 규칙(그 폴더는 시험 길이라 따로 둠). 마스터 파일은 읽기만 · 로그인 사용자 모두. +기계표 = M01 + 지금 층 내 기계(821-1 · `M02_MyMachine` — 식은 채운 칸으로 고른 같은 GC 식 · 직접 = 그 값) · +층을 모르는 길(`/api/m02/logic-lib/machine-hourly`)은 M01 만. """ from __future__ import annotations @@ -12,6 +14,7 @@ from __future__ import annotations from typing import Any from M01_MasterData import M01_MasterData_Store as store +from M02_MasterTemplete import M02_MyMachine as mym cm = store.cm _PLAIN = {"기계", "지역", "보정작업", "조종원직종"} # 이 밖의 입력 = 부착 장비 전용 @@ -43,7 +46,7 @@ def _rank(row: dict) -> tuple: def _operated(code: str) -> bool: """기계표(원문 8-4) 조종원 > 0 — 그 원문번호 기계 가운데 하나라도.""" - rows = store.loaded()[1].index.get("EQ", {}).values() + rows = mym.whole().index.get("EQ", {}).values() return any( str(r.get("원문번호")) == code and float((r.get("값") or {}).get("조종원") or 0) > 0 for r in rows @@ -56,7 +59,7 @@ def logic_of(code: str) -> dict | None: (「조종원 없는 기종」 목록에 겹친 기계 — 진동롤러(핸드가이드식) · 고소작업차 …).""" fit = [r for r in _family() if code in [str(c) for c in _codes(r)]] if not fit: - return _own(code) + return _own(code) or mym.logic_for(code) op = _operated(code) def order(row: dict) -> tuple: @@ -70,7 +73,7 @@ def logic_of(code: str) -> dict | None: def _own(code: str) -> dict | None: """그 기계만의 시간당 경비 식(816-1 · 4륜구동 1톤 덤프트럭 = GF000145 · 816-3 체인톱 = GF000337) — 「기계」 입력 없는 기계경비식이 그 원문번호 기계 줄 값(`EQ….가격` · `EQ….손료계수`)을 읽으면 그 식.""" - rows = store.loaded()[1].index.get("EQ", {}).values() + rows = mym.whole().index.get("EQ", {}).values() keys = [f"{r.get('키')}." for r in rows if code and str(r.get("원문번호")) == code] for data in store.loaded()[0].values(): for row in data.get("줄") or [] if data.get("그룹") == "로직" else []: @@ -91,9 +94,11 @@ def hourly(key: str) -> dict[str, Any]: if not key.startswith("EQ"): return {"ok": False, "reason": f"「{key}」 는 기계 아님"} try: - row = store.loaded()[1].get(key) + row = (whole := mym.whole()).get(key) except cm.mf.FormulaError as e: return {"ok": False, "reason": str(e)} + if mym.mine(key): + return mym.hourly(row, whole) code = str(row.get("원문번호") or "") logic = logic_of(code) if not logic: diff --git a/M02_MasterTemplete/M02_MyMachine.py b/M02_MasterTemplete/M02_MyMachine.py new file mode 100644 index 000000000..d7ec59cac --- /dev/null +++ b/M02_MasterTemplete/M02_MyMachine.py @@ -0,0 +1,363 @@ +"""M02 내 기계(PLAN 821-1 · 요구 795 · 요구서_796 3 · 7 장) — 개인 층 기계 줄 목록 · 서버 단독. + +저장 = 층 뿌리 `내기계.json` {줄: [줄], 다음: 수} — 줄 = M01 기계 줄 꼴 + {키 `EQP{사용자}-####`, 원문번호 「개인 {키}」, 원문 · 구분 · 출처 = 개인, 상세구분 · 이름, 규격, + 단위, 값, 비고, 방식 바로|품셈|직접, 시간당?(직접 {노무비, 재료비, 경비}), 넣은날, 고친날, 본뜬?} + 값 칸 이름 = M01 기계 줄과 같음 · 바로 = 손료계수 한 칸 · 품셈 = 비율 다섯 → 계수 넷(저장 때 서버가 + 계수 짝 `master_machine_coef` 로 다시 셈) · 직접 = 시간당 값만. +셈 색인 `whole()` = 지금 층(`store.root()`) 줄을 M01 위에 얹은 `skc.Copy`(키 앞 두 글자 EQ · + 원문번호로도 찾음) — 바로 = 상각비계수 ← 손료계수 · 정비 · 관리 0(식은 M01 기계와 같은 GC 식) · + 내 기계 없는 층 = M01 그대로. 단가산출 사본 · [단가 갱신] · 「단가새값」(`M02_Sangeun_Copy`) · + 기계 찾기(`M02_Machine_Find`) · 시간당(`M02_Machine_Hourly`) · 사본 모으기(`M02_Item_Prices.collect`)가 + 이 색인을 읽음 — 층을 모르는 길(`/api/m02/logic-lib/machine-hourly` 등)은 M01 만. +식 고름(채운 칸) = 연료 · 조종원 둘 다 GC000268 · 연료만 GC000272 · 조종원만 GC000276 · 없음 GC000273. +전력 칸은 담기만(M01 기계경비 식에 전력 줄 없음). +권한(본인 개인 층 · 맡은 회사 관리자)은 길(`M02_MyMachine_Router`) 몫. +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime +from decimal import Decimal +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json +from M01_MasterData import M01_MasterData_Store as m01 +from M02_MasterTemplete import M02_MasterTemplete_Store as store + +FILE = "내기계.json" +PREFIX, NO = "EQP", "개인 " +FIND = re.compile(rf"{NO}({PREFIX}\d+-\d+)") # 문서 안 기계 입력 「개인 EQP…」 +EASY, BOOK, DIRECT = "바로", "품셈", "직접" +COSTS = ("노무비", "재료비", "경비") +RATIOS = ("내용시간", "연간표준가동시간", "상각비율", "정비비율", "연간관리비율") +COEFS = ("상각비계수", "정비비계수", "관리비계수") +NUMS = ("가격", *RATIOS, *COEFS, "손료계수", "주연료", "잡재료율", "조종원", "전력") +UNITS = { + "가격": "천원", "내용시간": "시간", "연간표준가동시간": "시간", "상각비율": "", "정비비율": "", + "연간관리비율": "", **{k: "10^-7/시간" for k in (*COEFS, "손료계수")}, "주연료": "ℓ/hr", + "주연료종류": "", "잡재료율": "주연료의 %", "조종원": "인/일", "조종원직종": "", "전력": "kWh/hr", +} # fmt: skip +LOGIC = {(True, True): "GC000268", (True, False): "GC000272", + (False, True): "GC000276", (False, False): "GC000273"} # fmt: skip + + +def mine(key: Any) -> bool: + return str(key or "").startswith(PREFIX) + + +def _now() -> str: + return datetime.now().isoformat(timespec="seconds") + + +def _bad(text: str, status: int = 400) -> store.StoreError: + return store.StoreError(status, text) + + +# ── 저장 ────────────────────────────────────────────────────────────── +def _path(folder: Path | None = None) -> Path: + return Path(folder or store.root()) / FILE + + +def _read(folder: Path | None = None) -> dict[str, Any]: + try: + got = json.loads(_path(folder).read_bytes()) + except (OSError, ValueError): + got = {} + rows = [r for r in got.get("줄") or [] if isinstance(r, dict) and mine(r.get("키"))] + return {"줄": rows, "다음": int(got.get("다음") or 1)} + + +def rows(folder: Path | None = None) -> list[dict]: + """지금 층 내 기계 줄(저장 꼴).""" + return _read(folder)["줄"] + + +# ── 셈 색인 ────────────────────────────────────────────────────────── +def engine(row: dict) -> dict: + """셈 꼴 — 바로 = 상각비계수 ← 손료계수 · 정비 · 관리 0 · 빈 연료 · 조종원 칸 = 0 · 수 = Decimal.""" + vals = dict(row.get("값") or {}) + if row.get("방식") == EASY: + vals |= {"상각비계수": vals.get("손료계수") or 0, "정비비계수": 0, "관리비계수": 0} + if row.get("방식") != DIRECT: + vals = {"주연료": 0, "잡재료율": 0, "조종원": 0, "주연료종류": "", "조종원직종": ""} | vals + raw = json.dumps({**row, "값": vals}, ensure_ascii=False) + return json.loads(raw, parse_float=Decimal, parse_int=Decimal) + + +_MEMO: dict[str, tuple] = {} + + +def whole(base: Any = None) -> Any: + """M01 색인 + 지금 층 내 기계 줄 — 파일 · M01 이 그대로면 같은 객체(셈 메모가 붙음).""" + from M02_MasterTemplete import M02_Sangeun_Copy as skc + + base = m01.loaded()[1] if base is None else base + path = _path() + try: + st = path.stat() + except OSError: + return base + sig = (st.st_mtime_ns, st.st_size) + got = _MEMO.get(str(path)) + if got and got[0] == sig and got[1] is base: + return got[2] + own = [engine(r) for r in rows(path.parent)] + out = skc.Copy(base, own, ("내기계", sig)) if own else base + if len(_MEMO) > 16: # ponytail: 통째 비움 · 층이 많아지면 LRU 로 + _MEMO.clear() + _MEMO[str(path)] = (sig, base, out) + return out + + +def added(index: Any) -> list[dict]: + """`whole()` 이 얹은 내 기계 셈 줄(사본 모으기 `collect` 가 마스터 밖 줄로 담음).""" + return [r for k, r in (getattr(index, "index", {}).get("EQ") or {}).items() if mine(k)] + + +def keys_in(doc: Any) -> list[str]: + """문서 안 기계 입력 「개인 EQP…」 → 내 기계 키(줄 쌓기 `로직(GC…, 기계='개인 …')` · 사본 찍기).""" + return sorted(set(FIND.findall(json.dumps(doc, ensure_ascii=False, default=str)))) + + +def logic_key(row: dict) -> str: + vals = row.get("값") or {} + return LOGIC[(float(vals.get("주연료") or 0) > 0, float(vals.get("조종원") or 0) > 0)] + + +def logic_for(code: str) -> dict | None: + """내 기계 원문번호 → 시간당 경비 식(채운 칸으로 고름) · 내 기계 아님 = None.""" + index = whole() + row = (index.numbers.get("EQ") or {}).get(code) + if row is None or not mine(row.get("키")) or row.get("방식") == DIRECT: + return None + return index.logic(logic_key(row)) + + +def hourly(row: dict, index: Any) -> dict[str, Any]: + """`M02_Machine_Hourly.hourly` 꼴 — 직접 = 그 값 · 아니면 고른 GC 식(입력 기본값 · 기계 = 원문번호).""" + head = {"이름": row.get("이름"), "규격": row.get("규격"), "원문번호": row.get("원문번호")} + if row.get("방식") == DIRECT: + sums = {k: float((row.get("시간당") or {}).get(k) or 0) for k in COSTS} + return {"ok": True, **head, "로직": None, "직접": True, **sums, "계": sum(sums.values())} + import master_cut as mcut + + mf = m01.cm.mf + logic = index.logic(logic_key(row)) + given = {i["이름"]: (i.get("고르기") or [""])[0] for i in logic.get("입력") or []} + given["기계"] = str(row.get("원문번호")) + try: + got = mcut.engine( + index, logic["키"], mf.run(index, logic["키"], mf.fill_empty(logic, given)) + ) + except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e: + return {"ok": False, "reason": str(e) or type(e).__name__} + sums = {k: float(got[k]) for k in (*COSTS, "계")} + return {"ok": True, **head, "로직": logic["키"], **sums} + + +# ── 검사 ────────────────────────────────────────────────────────────── +def _num(value: Any, name: str) -> float | None: + if value in (None, ""): + return None + try: + got = float(value) + except (TypeError, ValueError): + raise _bad(f"「{name}」 은 수") from None + if got < 0 or got != got: + raise _bad(f"「{name}」 은 0 이상") + return int(got) if got == int(got) else got + + +def _coef(vals: dict) -> dict: + """품셈 방식 — 비율 다섯 → 계수 넷(계수 짝 `master_machine_coef` · 821-2).""" + import master_machine_coef as mmc + + got = mmc.coef_of(vals) + if got is None: + raise _bad("품셈 방식 — 내용시간 · 연간표준가동시간은 0 보다 큼") + return got + + +def clean(raw: Any) -> dict[str, Any]: + """받은 줄 → 저장 꼴 몸(키 · 날 빼고) — 틀리면 400.""" + if not isinstance(raw, dict): + raise _bad("내 기계 줄이 묶음 아님") + name, spec = str(raw.get("이름") or "").strip(), str(raw.get("규격") or "").strip() + if not name: + raise _bad("이름 없음") + how = raw.get("방식") or EASY + if how not in (EASY, BOOK, DIRECT): + raise _bad("방식은 바로 · 품셈 · 직접") + given = raw.get("값") if isinstance(raw.get("값"), dict) else {} + out: dict[str, Any] = {"이름": name, "규격": spec, "방식": how} + if how == DIRECT: + hour = raw.get("시간당") if isinstance(raw.get("시간당"), dict) else {} + out["시간당"] = {k: _num(hour.get(k), k) or 0 for k in COSTS} + vals: dict[str, Any] = {} + else: + vals = {k: got for k in NUMS if (got := _num(given.get(k), k)) is not None} + need = ("가격", "손료계수") if how == EASY else ("가격", *RATIOS) + if lost := [k for k in need if not vals.get(k)]: + raise _bad(f"빈 칸 — {' · '.join(lost)}") + drop = (*RATIOS, *COEFS) if how == EASY else () + vals = {k: v for k, v in vals.items() if k not in drop} + if how == BOOK: + vals |= _coef(vals) + vals |= _kinds(given, vals) + out["값"] = vals + out["단위"] = {k: UNITS[k] for k in vals if k in UNITS} + out["출처"] = {k: "개인" for k in vals} + if note := str(raw.get("비고") or "").strip(): + out["비고"] = note + return out + + +def _kinds(given: dict, vals: dict) -> dict: + """주연료종류 = M01 유가 이름 · 조종원직종 = M01 인력 원문번호(쓸 때만 · 고르기 목록 밖 = 400).""" + index, out = m01.loaded()[1], {} + fuel, job = str(given.get("주연료종류") or ""), str(given.get("조종원직종") or "") + if vals.get("주연료"): + if f"{fuel}:전국평균" not in (index.numbers.get("MO") or {}): + raise _bad(f"주연료종류 「{fuel}」 — 유가 목록에서 고름") + out["주연료종류"] = fuel + if vals.get("조종원"): + if job not in (index.numbers.get("LB") or {}): + raise _bad(f"조종원직종 「{job}」 — 인력 목록에서 고름") + out["조종원직종"] = job + return out + + +# ── 만들기 · 고치기 · 지우기 · 읽기 ───────────────────────────────────── +def _seed(key: str) -> dict: + """「비슷한 기계에서 시작」 — M01 · 내 기계 줄 값 복사(품셈 비율 있으면 품셈 · 없으면 바로).""" + try: + src = whole().get(key) + except m01.cm.mf.FormulaError: + raise _bad(f"없는 기계 「{key}」", 404) from None + if not str(src.get("키")).startswith("EQ"): + raise _bad(f"「{key}」 는 기계 아님") + if mine(key): + got = next(r for r in rows() if r["키"] == key) + return {k: got[k] for k in ("이름", "규격", "방식", "값", "시간당", "비고") if k in got} + vals = json.loads(json.dumps(src.get("값") or {}, default=float)) + how = BOOK if all(vals.get(k) for k in RATIOS) else EASY + return {"이름": src.get("이름"), "규격": src.get("규격"), "방식": how, "값": vals} + + +def _user() -> str: + """개인 층(`storage/{회사}/{사용자}/templates`) 사용자 번호 — 그 밖 층 = 404.""" + from M02_MasterTemplete import M02_Template_Layers as layers + + here = store.root().resolve() + user, company = here.parent.name, here.parent.parent.name + if not (user.isdigit() and company.isdigit()) or here != layers.personal_dir(company, user): + raise _bad("내 기계는 개인 층만", 404) + return user + + +def create(raw: Any = None, base: str | None = None) -> dict[str, Any]: + _user() + seed = _seed(base) if base else {} + raw = raw if isinstance(raw, dict) else {} + body = clean({**seed, **raw, "값": {**seed.get("값", {}), **(raw.get("값") or {})}}) + data = _read() + key = f"{PREFIX}{_user()}-{data['다음']:04d}" + when = _now() + row = {"키": key, "원문번호": f"{NO}{key}", "원문": "개인", "구분": "개인", "상세구분": body["이름"], + **body, "넣은날": when, "고친날": when, **({"본뜬": base} if base else {})} # fmt: skip + atomic_write_json(_path(), {"줄": [*data["줄"], row], "다음": data["다음"] + 1}) + return answer(row) + + +def update(key: str, raw: Any) -> dict[str, Any]: + _user() + data = _read() + old = next((r for r in data["줄"] if r["키"] == key), None) + if old is None: + raise _bad(f"없는 내 기계 「{key}」", 404) + body = clean(raw) + keep = {k: old[k] for k in ("키", "원문번호", "원문", "구분", "넣은날", "본뜬") if k in old} + row = {**keep, "상세구분": body["이름"], **body, "고친날": _now()} + atomic_write_json(_path(), {**data, "줄": [row if r["키"] == key else r for r in data["줄"]]}) + return answer(row) + + +def delete(key: str) -> dict[str, Any]: + """지움 — 이미 만든 단가산출은 사본의 마지막 값 그대로(「지운키」).""" + _user() + data = _read() + if not any(r["키"] == key for r in data["줄"]): + raise _bad(f"없는 내 기계 「{key}」", 404) + used = uses().get(key, []) + atomic_write_json(_path(), {**data, "줄": [r for r in data["줄"] if r["키"] != key]}) + return {"지움": key, "쓰는곳": used} + + +def uses() -> dict[str, list[dict]]: + """{내 기계 키: [{키, 이름}]} — 이 층 단가산출 가운데 사본 단가 줄에 그 키를 가진 것.""" + from M02_MasterTemplete import M02_Sangeun_Copy as skc + from M02_MasterTemplete import M02_Sangeun_Doc as sd + + out: dict[str, list[dict]] = {} + for path in sorted((store.root() / sd.KIND).glob("SK-*.json")): + raw = path.read_bytes() + if PREFIX.encode() not in raw: + continue + try: + doc = json.loads(raw) + except ValueError: + continue + for row in (doc.get(skc.KEY) or {}).get("단가") or []: + if mine(row.get("키")): + out.setdefault(str(row["키"]), []).append( + {"키": doc.get("키"), "이름": doc.get("이름")} + ) + return out + + +def same(name: Any, spec: Any, skip: str = "") -> list[dict]: + """같은 이름 · 규격 알림 — M01 기계 · 이 층 내 기계(표기 정규화 `M02_Machine_Find.norm`).""" + from M02_MasterTemplete.M02_Machine_Find import norm + + want = (norm(name), norm(spec)) + return [ + { + "키": r["키"], + "이름": r.get("이름"), + "규격": r.get("규격"), + "출처": "개인" if mine(r["키"]) else "M01", + } # fmt: skip + for r in (whole().index.get("EQ") or {}).values() + if r["키"] != skip and (norm(r.get("이름")), norm(r.get("규격"))) == want + ] + + +def answer(row: dict, used: list | None = None) -> dict[str, Any]: + out = { + "줄": row, + "시간당": hourly(engine(row), whole()), + "같은이름": same(row["이름"], row["규격"], row["키"]), + } + return out | ({"쓰는곳": used} if used is not None else {}) + + +def one(key: str) -> dict[str, Any]: + got = next((r for r in rows() if r["키"] == key), None) + if got is None: + raise _bad(f"없는 내 기계 「{key}」", 404) + return answer(got, uses().get(key, [])) + + +def listing() -> dict[str, Any]: + """목록 — 줄마다 시간당 · 쓰는곳수(넣은 차례).""" + index, used = whole(), uses() + return { + "줄": [ + {**r, "시간당": hourly(engine(r), index), "쓰는곳수": len(used.get(r["키"], []))} + for r in rows() + ] + } diff --git a/M02_MasterTemplete/M02_MyMachine_Router.py b/M02_MasterTemplete/M02_MyMachine_Router.py new file mode 100644 index 000000000..33b83ad18 --- /dev/null +++ b/M02_MasterTemplete/M02_MyMachine_Router.py @@ -0,0 +1,73 @@ +"""M02 내 기계 길(PLAN 821-1 · `M02_MyMachine`) — `/api/m02/layers/personal/my-machines`(?owner=). + +GET 목록(줄마다 시간당 · 쓰는곳수) · GET `/{키}` 하나(+시간당 · 쓰는곳 · 같은이름) · +POST 만들기 몸 {줄?, 본뜬?: M01 · 내 기계 키(「비슷한 기계에서 시작」 · 「M01 기계 복사해 만들기」)} · +PUT `/{키}` 몸 {줄} · DELETE `/{키}`(답 = 쓰는곳 — 그 단가산출은 사본의 마지막 값 그대로). +권한 = 층 권한 한 벌(`Router_Items._root` — 본인 개인 층 · 맡은 회사 관리자) · 개인 층만(그 밖 404). +⚠ 등록은 `main.py`(로그인만). +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from common_util.common_util_auth import verify_session +from M02_MasterTemplete import M02_MyMachine as mym +from M02_MasterTemplete.M02_MasterTemplete_Router_Items import Session, _call, _root + +router = APIRouter(prefix="/api/m02", tags=["M02 My Machine"]) +PATH = "/layers/{layer}/my-machines" + + +class MachineBody(BaseModel): + 줄: dict[str, Any] | None = None + 본뜬: str | None = None + + +def _folder(session: Session, layer: str, owner: str | None, write: bool = False): + if layer != "personal": + raise HTTPException(status_code=404, detail="내 기계는 개인 층만") + return _root(session, layer, owner, write=write) + + +@router.get(PATH) +def my_machines(layer: str, owner: str | None = None, session: Session = Depends(verify_session)): + return _call(_folder(session, layer, owner), mym.listing) + + +@router.get(PATH + "/{key}") +def my_machine( + layer: str, key: str, owner: str | None = None, session: Session = Depends(verify_session) +): + return _call(_folder(session, layer, owner), mym.one, key) + + +@router.post(PATH) +def my_machine_create( + layer: str, + body: MachineBody, + owner: str | None = None, + session: Session = Depends(verify_session), +): + return _call(_folder(session, layer, owner, True), mym.create, body.줄, body.본뜬) + + +@router.put(PATH + "/{key}") +def my_machine_update( + layer: str, + key: str, + body: MachineBody, + owner: str | None = None, + session: Session = Depends(verify_session), +): + return _call(_folder(session, layer, owner, True), mym.update, key, body.줄) + + +@router.delete(PATH + "/{key}") +def my_machine_delete( + layer: str, key: str, owner: str | None = None, session: Session = Depends(verify_session) +): + return _call(_folder(session, layer, owner, True), mym.delete, key) diff --git a/M02_MasterTemplete/M02_Sangeun_Calc.py b/M02_MasterTemplete/M02_Sangeun_Calc.py index cb4f6d15d..b32a1a6e7 100644 --- a/M02_MasterTemplete/M02_Sangeun_Calc.py +++ b/M02_MasterTemplete/M02_Sangeun_Calc.py @@ -549,10 +549,8 @@ def _check_refs(doc: dict, here_: refs.Here) -> None: def clean_doc(doc: Any, here_: refs.Here) -> dict[str, Any]: """저장 꼴(`Labor_Ref.clean_doc` 가 부름) — 검사 · 참조 · 순환 · 서버가 `원값` · `바탕판`.""" - from M01_MasterData import M01_MasterData_Store as m01 - raw, job = doc, doc.get(skc.JOB) if isinstance(doc, dict) else None # 수식 사본(802-4a) - whole = skc.over(doc := skc.settle(doc, True), m01.loaded()[1]) # 사본(795-1) + whole = skc.over(doc := skc.settle(doc, True), skc.whole_()) # 사본(795-1) · 내 기계(821-1) try: rough = sd.clean(doc, None, whole) _check_refs(rough, here_) @@ -604,7 +602,7 @@ def calc_doc(doc: Any, given: dict | None = None, seen: bool = True) -> dict[str from M01_MasterData import M01_MasterData_Store as m01 with m01.cache.steady(): # 한 셈 동안 M01 파일 자국 한 번(792-2) - whole = skc.over(doc := skc.settle(doc), m01.loaded()[1]) # 사본(795-1) + whole = skc.over(doc := skc.settle(doc), skc.whole_()) # 사본(795-1) · 내 기계(821-1) try: clean = sd.clean(doc, None, whole) _check_refs(clean, refs.here()) # 저장 전 문서도 위 층만 · 있는 문서 · 순환 422 diff --git a/M02_MasterTemplete/M02_Sangeun_Copy.py b/M02_MasterTemplete/M02_Sangeun_Copy.py index 0d01e49e9..950b41bac 100644 --- a/M02_MasterTemplete/M02_Sangeun_Copy.py +++ b/M02_MasterTemplete/M02_Sangeun_Copy.py @@ -30,6 +30,7 @@ from decimal import Decimal from typing import Any from M02_MasterTemplete import M02_Item_Prices as prices +from M02_MasterTemplete import M02_MyMachine as mym from M02_MasterTemplete import M02_Sangeun_Doc as sd from M02_MasterTemplete import M02_Sangeun_Imported as imp @@ -41,7 +42,10 @@ _MASTER = prices.mf.Master def whole_() -> Any: - return prices.m01.loaded()[1] + """M01 색인 + 지금 층 내 기계 줄(821-1 · `M02_MyMachine.whole`).""" + from M02_MasterTemplete import M02_MyMachine as mym + + return mym.whole() def _fp(rows: Any) -> str: @@ -149,6 +153,7 @@ def stamp(doc: dict) -> dict | None: return None keys, links = reach(doc.get("더한줄")) keys = [str(base["로직"]), *keys] if base.get("로직") else keys + links += mym.keys_in(doc) # 줄 쌓기 기계 입력 「개인 EQP…」(821-1) extra = [row] if (row := imp.row_of(doc)) else [] return snap(keys, links, extra) diff --git a/main.py b/main.py index ec057c6bb..9f5954106 100644 --- a/main.py +++ b/main.py @@ -578,6 +578,11 @@ app.include_router( __import__("M02_MasterTemplete.M02_Sangeun_Router", fromlist=["router"]).router, dependencies=protected, ) +# 내 기계(PLAN 821-1) — 로그인 · 개인 층 권한은 길 안에서 +app.include_router( + __import__("M02_MasterTemplete.M02_MyMachine_Router", fromlist=["router"]).router, + dependencies=protected, +) # 단가산출(SK) 수식 문서 길(PLAN 802-4a) — 로그인 · 층 권한은 길 안에서 app.include_router( __import__("M02_MasterTemplete.M02_Doc_Router", fromlist=["router"]).router, diff --git a/resources/tester/test_821_1_my_machine.py b/resources/tester/test_821_1_my_machine.py new file mode 100644 index 000000000..98a2980de --- /dev/null +++ b/resources/tester/test_821_1_my_machine.py @@ -0,0 +1,178 @@ +# ruff: noqa: F811 — 임시 층 fixture(world)를 test_m02_labor_ref 에서 받아 씀 +"""821-1 내 기계(개인 층 기계 줄) — M01 기계처럼: 만들 때 사본에 복사 · 고쳐도 금액 그대로 + 「단가새값」 · +[단가 갱신] 으로 반영 · 지워도 마지막 값(임시 storage · M01 정본 안 고침). + +① 세 꼴(바로 · 품셈 · 직접) 저장 · 키 · 원문번호 M01 과 안 겹침 ② 시간당 = M01 같은 값 기계와 같음 +③ 찾기 · 남의 층 안 보임 ④ SK 사본에 박힘 · 고친 뒤 그대로 + 새값 · 갱신 뒤 반영 · 지운 뒤 마지막 값 · +층 사이 복사 ⑤ M01 만 쓰는 SK 금액 그대로 ⑥ 권한. +""" + +from __future__ import annotations + +import json + +import pytest +from fastapi import HTTPException + +from M01_MasterData import M01_MasterData_Store as m01 +from M02_MasterTemplete import M02_Labor_Ref as refs +from M02_MasterTemplete import M02_Machine_Find as mfind +from M02_MasterTemplete import M02_Machine_Hourly as mh +from M02_MasterTemplete import M02_MasterTemplete_Store as store +from M02_MasterTemplete import M02_MyMachine as mym +from M02_MasterTemplete import M02_MyMachine_Router as route +from M02_MasterTemplete import M02_Sangeun_Calc as calc +from M02_MasterTemplete import M02_Sangeun_Copy as skc +from M02_MasterTemplete import M02_Template_Layers as layers +from resources.tester.test_m02_labor_ref import world # noqa: F401 — 같은 임시 층 + +DIG = "EQ000022" # 굴착기(무한궤도) 1.0㎥ — 연료 · 조종원 있음 +PCT = "GC001109" # M01 만 쓰는 SK 바탕 +NAME = "서브셋임시기계" # 찾기 기종 글(수 = 규격으로 읽음 · 창 표시는 글로) + + +def _at(folder, fn, *args): + with store.at(folder): + return fn(*args) + + +def _easy(**vals): + base = {"가격": 1000, "손료계수": 2000, "주연료": 2, "주연료종류": "경유", "잡재료율": 10, + "조종원": 1, "조종원직종": "1048"} # fmt: skip + return {"이름": NAME, "규격": "1㎾", "방식": "바로", "값": base | vals} + + +def _sk(folder, machine_no): + line = {"종류": "로직", "로직": "GC000268", "입력값": {"기계": machine_no, "지역": "전국평균"}, + "수량": "1"} # fmt: skip + body = {"이름": "sub3임시 내기계", "단위": "시간", "더한줄": [line]} + return _at(folder, lambda: refs.create("sangeun", body)["이름"]) + + +def _doc(folder, key): + return _at(folder, lambda: store.read("sangeun", key)["문서"]) + + +def _calc(folder, doc): + return _at(folder, calc.calc_doc, doc, None, False) + + +def _sum(folder, key): + return _calc(folder, _doc(folder, key))["기준"]["계"] + + +def test_three_shapes_and_keys(world): + """① 바로 · 품셈 · 직접 — 키 EQP{사용자}-#### · 원문번호 「개인 …」 · M01 과 안 겹침 · 날.""" + me = world["personal"] + easy = _at(me, mym.create, _easy())["줄"] + assert easy["키"] == "EQP42-0001" and easy["원문번호"] == "개인 EQP42-0001" + assert set(easy["값"]) >= {"가격", "손료계수"} and "상각비계수" not in easy["값"] + assert easy["넣은날"] == easy["고친날"] and easy["출처"]["가격"] == "개인" + book = _at(me, mym.create, {"방식": "품셈"}, DIG)["줄"] # M01 기계 복사해 만들기(비율 → 계수) + m01_row = m01.loaded()[1].get(DIG)["값"] + for k in ("상각비계수", "정비비계수", "관리비계수", "손료계수"): + assert float(book["값"][k]) == float(m01_row[k]), k + direct = _at(me, mym.create, {"이름": NAME, "방식": "직접", + "시간당": {"노무비": 10, "재료비": 20, "경비": 30}}) # fmt: skip + assert direct["시간당"]["계"] == 60 and direct["시간당"]["직접"] is True + m01_eq = m01.loaded()[1] + for row in _at(me, mym.rows): + assert row["키"] not in m01_eq.index["EQ"] and row["원문번호"] not in m01_eq.numbers["EQ"] + assert [r["키"] for r in _at(me, mym.rows)] == ["EQP42-0001", "EQP42-0002", "EQP42-0003"] + assert {x["키"] for x in direct["같은이름"]} == {"EQP42-0001"} # 같은 이름 · 규격 알림 + for bad in ({"이름": ""}, {**_easy(), "값": {"가격": 1}}, {**_easy(가격=-1)}, + {**_easy(주연료종류="물")}, {**_easy(조종원직종="없음")}): # fmt: skip + with pytest.raises(store.StoreError): + _at(me, mym.create, bad) + + +def test_hourly_same_as_m01(world): + """② 굴착기 1.0㎥ 값을 내 기계로(바로 · 품셈) → 시간당 = M01 기계 · 식은 채운 칸으로.""" + me = world["personal"] + want = mh.hourly(DIG) + for how in ("바로", "품셈"): + got = _at(me, mym.create, {"방식": how}, DIG) + hour = _at(me, mh.hourly, got["줄"]["키"]) + assert hour["ok"] and hour["로직"] == "GC000268", how + for k in ("노무비", "재료비", "경비", "계"): + assert hour[k] == want[k], (how, k) + loss = _at(me, mym.create, _easy(주연료=0, 조종원=0))["시간당"] + assert loss["로직"] == "GC000273" and loss["계"] == loss["경비"] == 200 # 1000×1000×2000/10⁷ + assert _at(me, mym.create, _easy(조종원=0))["시간당"]["로직"] == "GC000272" + assert _at(me, mym.create, _easy(주연료=0))["시간당"]["로직"] == "GC000276" + + +def test_find_and_other_layer_hidden(world): + """③ 기계 찾기 = 그 층 내 기계도 · 남의 층 · 층 모르는 길 = M01 만.""" + me, other = world["personal"], layers.personal_dir(7, 43) + key = _at(me, mym.create, _easy())["줄"]["키"] + got = _at(me, mfind.find, NAME, "1㎾") + assert got["판정"] == mfind.LINKED and got["키"] == key + assert _at(other, mfind.find, NAME, "1㎾")["판정"] == mfind.NONE + assert _at(other, mh.hourly, key)["ok"] is False + assert mh.hourly(key)["ok"] is False # 층 모르는 길 + assert key not in _at(other, skc.whole_).index["EQ"] + + +def test_sk_copy_edit_refresh_delete(world): + """④ 만들 때 사본에 박힘 → 고쳐도 그대로 + 새값 → [단가 갱신] 반영 → 지워도 마지막 값 · 층 복사.""" + me = world["personal"] + made = _at(me, mym.create, _easy()) + key, no = made["줄"]["키"], made["줄"]["원문번호"] + sk = _sk(me, no) + copy = _doc(me, sk)["사본"] + assert key in {r["키"] for r in copy["단가"]} + was = _sum(me, sk) + assert float(was) == made["시간당"]["계"] + assert _at(me, mym.one, key)["쓰는곳"] == [{"키": sk, "이름": "sub3임시 내기계"}] + _at(me, mym.update, key, _easy(가격=3000)) # 고침 + now = _calc(me, _doc(me, sk)) + assert now["기준"]["계"] == was and now["사본"]["단가새값"] is True + assert now["사본"]["지운키"] == [] + new = _at(me, mh.hourly, key)["계"] + look = _at(me, skc.preview, _doc(me, sk), skc.PRICE) + assert float(look["전"]["계"]) == float(was) and float(look["후"]["계"]) == new + assert [x["키"] for x in look["바뀐자원"]] == [key] + with store.at(me): # [저장] = 할일 적용 + store.write("sangeun", sk, store.read("sangeun", sk)["판"], look["문서"]) + assert float(_sum(me, sk)) == new + assert _calc(me, _doc(me, sk))["사본"]["단가새값"] is False + # 층 사이 복사 — 사본이 따라감(받은 층에 그 내 기계 없음 · 마지막 값) + other = layers.personal_dir(7, 43) + moved = _at(other, skc.settle, {**_doc(me, sk), "키": "SK-999999"}) + assert float(_calc(other, moved)["기준"]["계"]) == new + gone = _at(me, mym.delete, key) + assert gone["쓰는곳"] == [{"키": sk, "이름": "sub3임시 내기계"}] + after = _calc(me, _doc(me, sk)) + assert float(after["기준"]["계"]) == new and after["사본"]["지운키"] == [key] + look = _at(me, skc.preview, _doc(me, sk), skc.PRICE) + assert look["바뀐자원"] == [{"키": key, "이름": NAME, "지워짐": True}] + assert float(look["후"]["계"]) == new + + +def test_m01_only_sk_same(world): + """⑤ 내 기계가 있는 층에서도 M01 만 쓰는 SK 금액 · 사본 같음 · 사본 모으기에 내 기계 안 낌.""" + me, other = world["personal"], layers.personal_dir(7, 43) + base = {"이름": "sub3임시 M01", "단위": "㎡", "바탕": {"로직": PCT}, "더한줄": []} + plain = _at(other, lambda: refs.create("sangeun", base)["이름"]) + _at(me, mym.create, _easy()) + mine = _at(me, lambda: refs.create("sangeun", base)["이름"]) + a, b = _doc(other, plain)["사본"], _doc(me, mine)["사본"] + assert json.dumps(a["단가"], sort_keys=True) == json.dumps(b["단가"], sort_keys=True) + assert a["판"] == b["판"] and _sum(other, plain) == _sum(me, mine) + + +def test_rights(world): + """⑥ 남의 개인 층 = 403(일반) · 회사 관리자 = 됨 · 개인 층 밖 = 404.""" + user = {"role": "USER", "company_id": 7, "user_id": 42} + boss = {"role": "ADMIN", "company_id": 7, "user_id": 1} + with pytest.raises(HTTPException) as e: + route._folder(user, "personal", "7/43", True) + assert e.value.status_code == 403 + with pytest.raises(HTTPException) as e: + route._folder(user, "company", None, True) + assert e.value.status_code == 404 + assert route._folder(user, "personal", None, True) == world["personal"] + assert route._folder(boss, "personal", "7/42") == world["personal"] # 맡은 회사 관리자 읽기 + with pytest.raises(store.StoreError): # 개인 층 아닌 자리에서 만들기 + _at(world["company"], mym.create, _easy())