Aislo/M02_MasterTemplete/StructOptions/M02_SO_Groups.py

"""M02 구조물 옵션 그룹 — 구조물 종류 안에 이름 붙은 묶음(PLAN 46-2).

정본 = `resources/master_template/구조물옵션그룹.json` =
    `{type_id: [{"id": "g1a2b3c", "이름": "유입구 기슭막이", "옵션": ["inlet_revet_form", …]}]}`
— 없는 종류 = 그룹 없음(파일이 없어도 됨) · 빈 그룹(옵션 0)도 그룹.

그룹 하나 = 프로젝트 B05 · B06 폼의 보조 컨테이너 하나(제목 = 이름) — 그 안에서 구조물 조합
항목 하나를 지정(구조물 기록 `group_items[그룹 id]`). 그룹에 안 넣은 옵션 = 본체 칸(`summary_item`).
옵션 하나는 한 종류 안에서 그룹 하나에만. 시스템 공통 · 고쳐 씀은 시스템 관리자(길이 막음).
판 = 파일 sha256 앞 16자(`store.version_of`) · 낡은 판으로 쓰면 409.
"""

from __future__ import annotations

import json
import logging
import re
from typing import Any

from B05_Profile.B05_Profile_Structures_Schema import load_structure_types
from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_MasterTemplete_Store as store

FILE = "구조물옵션그룹.json"
GROUP_ID = re.compile(r"^g[0-9a-z]{1,16}$")
NAME_MAX = 40
logger = logging.getLogger(__name__)


def _load() -> tuple[dict[str, Any], str]:
    """(그룹 표 · 판) — 파일이 없으면 빈 표 · 판 빈 글."""
    path = store.FOLDER / FILE
    if not path.is_file():
        return {}, ""
    raw = path.read_bytes()
    try:
        doc = json.loads(raw)
    except ValueError as e:
        raise store.StoreError(500, f"{FILE} 을 못 읽음 — {e}") from e
    if not isinstance(doc, dict):
        raise store.StoreError(500, f"{FILE} 은 종류별 묶음이어야 함")
    return doc, store.version_of(raw)


def groups_of() -> dict[str, list[dict[str, Any]]]:
    """종류 → `[{id, name, options}]` — 폼 응답용 · 깨진 표 · 모르는 종류는 빼고 로그만."""
    try:
        doc = _load()[0]
    except store.StoreError as e:
        logger.warning("구조물 옵션 그룹 표를 못 읽어 그룹 없음으로 봄 — %s", e.detail)
        return {}
    known = {t.type_id: {o.key for o in t.options} for t in load_structure_types()}
    out: dict[str, list[dict[str, Any]]] = {}
    for type_id, rows in doc.items():
        if type_id not in known or not isinstance(rows, list):
            continue
        for row in rows:
            if not isinstance(row, dict) or not GROUP_ID.match(str(row.get("id", ""))):
                continue
            keys = [k for k in row.get("옵션") or [] if k in known[type_id]]
            out.setdefault(type_id, []).append(
                {"id": row["id"], "name": str(row.get("이름") or row["id"]), "options": keys}
            )
    return out


def listing() -> dict[str, Any]:
    """`{판, 그룹, 종류}` — 종류 = 레지스트리 전부(옵션 없는 종류도 · 빈 그룹을 둘 수 있음)."""
    doc, version = _load()
    kinds = [
        {
            "type_id": t.type_id,
            "group": t.group,
            "name": t.name,
            "enabled": t.enabled,
            "options": [
                {"key": o.key, "label": o.label, "phase": o.phase, "required": o.required}
                for o in t.options
            ],
        }
        for t in load_structure_types()
    ]
    return {"판": version, "그룹": doc, "종류": kinds}


def _clean(groups: Any) -> dict[str, list[dict[str, Any]]]:
    """받은 표 검사 — 없는 종류 · 키 · 겹친 id · 이름 · 옵션 두 그룹은 422 · 빈 종류는 뺌."""
    if not isinstance(groups, dict):
        raise store.StoreError(400, "그룹은 종류별 묶음이어야 함")
    keys = {t.type_id: {o.key for o in t.options} for t in load_structure_types()}
    errors: list[str] = []
    out: dict[str, list[dict[str, Any]]] = {}
    for type_id, rows in groups.items():
        if type_id not in keys:
            errors.append(f"없는 구조물 종류 「{type_id}」")
            continue
        if not isinstance(rows, list):
            errors.append(f"「{type_id}」 는 그룹 목록이어야 함")
            continue
        ids: set[str] = set()
        names: set[str] = set()
        owner: dict[str, str] = {}
        clean: list[dict[str, Any]] = []
        for row in rows:
            if not isinstance(row, dict):
                errors.append(f"「{type_id}」 그룹은 묶음이어야 함")
                continue
            gid = str(row.get("id", ""))
            name = str(row.get("이름", "")).strip()
            opts = row.get("옵션") or []
            if not GROUP_ID.match(gid) or gid in ids:
                errors.append(f"「{type_id}」 그룹 id 「{gid}」 가 틀리거나 겹침")
            if not name or len(name) > NAME_MAX or name in names:
                errors.append(
                    f"「{type_id}」 그룹 이름 「{name}」 은 1~{NAME_MAX}자 · 겹치면 안 됨"
                )
            if not isinstance(opts, list):
                errors.append(f"「{type_id}.{name}」 옵션은 목록이어야 함")
                opts = []
            for key in opts:
                if key not in keys[type_id]:
                    errors.append(f"「{type_id}」 에 없는 옵션 「{key}」")
                elif key in owner:
                    errors.append(
                        f"「{type_id}.{key}」 는 「{owner[key]}」 · 「{name}」 두 그룹에 있음"
                    )
                else:
                    owner[key] = name
            ids.add(gid)
            names.add(name)
            clean.append({"id": gid, "이름": name, "옵션": [k for k in opts if k in keys[type_id]]})
        if clean:
            out[type_id] = clean
    if errors:
        raise store.StoreError(422, {"errors": errors})
    return out


def save(groups: Any, version: str) -> dict[str, Any]:
    """고쳐 씀 — 판이 같을 때만. 돌려주는 것 = 새 `listing()`."""
    fixed = _clean(groups)
    path = store.FOLDER / FILE
    with store._LOCK:
        have = store.version_of(path.read_bytes()) if path.is_file() else ""
        if have != (version or ""):
            raise store.StoreError(409, {"stale": [FILE], "판": have})
        atomic_write_json(path, fixed)
    return listing()