Aislo/M02_MasterTemplete/M02_Item_Copy.py

"""구조물 집계 항목 복사본 읽기 · 쓰기 · 재적용 · 미리 가져오기 · 개인에 저장 · 새 항목 · 계산
(PLAN 25-2b · 26-2b).

복사본 = `{프로젝트}/templates/structures/{구조물 id}/`
(`M02_Item_Snapshot`) · 길의 폴더 = 구조물 id.
파일마다 판 = 내용 해시 — 쓸 때 판이 다르면 `Stale`(409). 고치면 manifest 에 `고침` 을 남겨 지울 때
`_지난/` 에 옮겨 둠. 옛 복사본(새 파일 없음) · 옛 항목 폴더는 읽을 때 채우고 옮겨 받음.
계산은 복사본만 씀(마스터 현재값 안 읽음) — 원본 항목이 바뀌거나 지워져도 그대로.
"""

from __future__ import annotations

import re
from collections.abc import Callable
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_Item_Personal as personal
from M02_MasterTemplete import M02_Item_Prices as prices
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_Item_Snapshot import (
    FOLDER_PATTERN,
    LOCAL_FILES,
    MANIFEST_NAME,
    PICKS_FILE,
    VALUES_FILE,
    build_snapshot,
    complete_old,
    folder_name,
    items_dir,
    migrate_items,
    now_text,
    retire,
    same_item,
    source_files,
    structures_dir,
    write_snapshot,
)
from M02_MasterTemplete.M02_Item_Source import (
    LABOR_FILE,
    LINK_FILE,
    PRICE_FILE,
    build_source,
    read_json,
    version_of,
)

#: 길의 종류 이름 → 복사본 파일
KINDS = {
    "도면": "도면.json",
    "면적": "면적.json",
    "재료": "재료.json",
    "자재조합": "자재조합.json",
    "입력연결": LINK_FILE,
    "입력값": VALUES_FILE,
    "고른자재": PICKS_FILE,
}
#: 뼈대 검사에 쓰는 층 양식 종류(도면 · 면적산출 · 재료산출)
SKELETON = {"도면": "structure", "면적": "area", "재료": "material"}
SEATS = ("면적", "재료", "선택", None)
_MT = re.compile(r"^MT\d{6}$")


class Stale(ValueError):
    def __init__(self, current: str):
        super().__init__("다른 곳에서 먼저 고쳤습니다 — 다시 읽고 고치세요.")
        self.current = current


def _migrated(project_root: str | Path) -> Path:
    """구조물 복사본 뿌리 — 옛 항목 폴더가 남았으면 먼저 옮겨 받음."""
    if items_dir(project_root).is_dir():
        migrate_items(project_root)
    return structures_dir(project_root)


def folder_of(project_root: str | Path, folder: str) -> Path:
    """구조물 복사본 폴더 — 이름 모양이 틀리거나 없으면 LookupError(404)."""
    if not FOLDER_PATTERN.match(folder):
        raise LookupError(f"복사본 폴더 이름 모양이 아님 「{folder}」")
    target = _migrated(project_root) / folder
    if not (target / MANIFEST_NAME).is_file():
        raise LookupError(f"복사본이 없음 「{folder}」")
    complete_old(target)
    return target


def source_changed(manifest: dict[str, Any]) -> bool | None:
    """지금 원본(층 · 마스터)을 다시 모은 판이 가져온 판과 다른가 — 원본이 없어졌으면 None."""
    try:
        source = build_source(manifest["원본"])
    except (LookupError, ValueError, KeyError):
        return None
    return version_of(source) != manifest.get("원본판")


def list_items(project_root: str | Path) -> list[dict[str, Any]]:
    base = _migrated(project_root)
    out = []
    for child in sorted(base.iterdir()) if base.is_dir() else []:
        if not child.is_dir() or not FOLDER_PATTERN.match(child.name):
            continue
        if not (child / MANIFEST_NAME).is_file():
            continue
        complete_old(child)
        manifest = read_json(child / MANIFEST_NAME) or {}
        origin = manifest.get("원본") or {}
        changed = source_changed(manifest)
        out.append(
            {
                "폴더": child.name,
                "이름": origin.get("name"),
                "원본": origin,
                "원본판": manifest.get("원본판"),
                "가져온날": manifest.get("가져온날"),
                "마스터바뀜": bool(changed),
                "원본없음": changed is None,
                "구조물": manifest.get("구조물") or [],
                "고침": bool(manifest.get("고침")),
            }
        )
    return out


def read_file(project_root: str | Path, folder: str, kind: str) -> dict[str, Any]:
    target = folder_of(project_root, folder)
    content = read_json(target / KINDS[kind])
    return {"종류": kind, "판": version_of(content), "내용": content}


# ── 모양 검사 ───────────────────────────────────────────────────────


def _check_values(content: Any) -> None:
    if not isinstance(content, dict):
        raise ValueError("입력값은 {로직키: {입력이름: {값, 확인}}} 이어야 합니다.")
    for logic, inputs in content.items():
        if not isinstance(inputs, dict):
            raise ValueError(f"입력값 「{logic}」 은 {{입력이름: {{값, 확인}}}} 이어야 합니다.")
        for name, one in inputs.items():
            if not isinstance(one, dict) or set(one) - {"값", "확인"}:
                raise ValueError(f"입력값 「{logic}|{name}」 은 {{값, 확인}} 이어야 합니다.")
            if not isinstance(one.get("확인", False), bool):
                raise ValueError(f"입력값 「{logic}|{name}」 확인은 참 · 거짓이어야 합니다.")
            if isinstance(one.get("값"), (dict, list)):
                raise ValueError(f"입력값 「{logic}|{name}」 값은 수 · 글 · 빈 값이어야 합니다.")


def _check_picks(content: Any) -> None:
    if not isinstance(content, dict):
        raise ValueError("고른자재는 {로직키: {호표 차례: MT키}} 이어야 합니다.")
    for logic, picks in content.items():
        if not isinstance(picks, dict):
            raise ValueError(f"고른자재 「{logic}」 은 {{호표 차례: MT키}} 이어야 합니다.")
        for at, key in picks.items():
            if not str(at).isdigit() or not isinstance(key, str) or not _MT.match(key):
                raise ValueError(f"고른자재 「{logic}」 {at} → 「{key}」 — 차례는 수 · 값은 MT 키")


def _check_links(content: Any) -> None:
    rows = content.get("줄") if isinstance(content, dict) else None
    if not isinstance(rows, list):
        raise ValueError("입력연결은 {판, 줄: [...]} 이어야 합니다.")
    for row in rows:
        if not isinstance(row, dict) or not row.get("로직") or not row.get("입력"):
            raise ValueError("입력연결 줄은 {로직, 입력, 자리, 알약?} 이어야 합니다.")
        if row.get("자리") not in SEATS:
            raise ValueError(f"입력연결 자리 「{row.get('자리')}」 — 면적 · 재료 · 선택 · 빈 값")


def check(kind: str, content: Any) -> None:
    if kind == "입력값":
        _check_values(content)
    elif kind == "고른자재":
        _check_picks(content)
    elif kind == "입력연결":
        _check_links(content)
    elif kind in SKELETON:
        layers.check_skeleton(SKELETON[kind], content)
    elif not isinstance(content, dict):
        raise ValueError(f"{kind} 은 JSON 객체여야 합니다.")


def write_file(
    project_root: str | Path,
    folder: str,
    kind: str,
    version: str | None,
    content: Any,
    recalc: Callable[[dict], dict] | None = None,
) -> dict[str, Any]:
    """판이 같을 때만 씀 — 면적 · 재료는 `recalc`(서버 계산값)를 거친 것을 씀."""
    target = folder_of(project_root, folder)
    check(kind, content)
    current = version_of(read_json(target / KINDS[kind]))
    if version != current:
        raise Stale(current)
    if recalc is not None and kind in ("면적", "재료"):
        content = recalc(content)
    atomic_write_json(target / KINDS[kind], content)
    manifest = read_json(target / MANIFEST_NAME) or {}
    atomic_write_json(target / MANIFEST_NAME, {**manifest, "고침": True, "고친날": now_text()})
    return {"종류": kind, "판": version_of(content)}


# ── 새로 가져오기 · 미리 가져오기 ─────────────────────────────────────


def _kept_values(values: Any, labor: dict[str, Any]) -> dict[str, Any]:
    """같은 로직 · 같은 입력 이름만 살림."""
    names = {
        str(row.get("키")): {spec["이름"] for spec in row.get("입력", [])}
        for row in labor.get("로직", [])
    }
    out: dict[str, Any] = {}
    for logic, inputs in (values or {}).items():
        kept = {k: v for k, v in (inputs or {}).items() if k in names.get(logic, set())}
        if kept:
            out[logic] = kept
    return out


def _kept_picks(picks: Any, labor: dict[str, Any]) -> dict[str, Any]:
    """같은 로직 · 있는 호표 차례만 살림."""
    lines = {str(row.get("키")): len(row.get("호표", [])) for row in labor.get("로직", [])}
    out: dict[str, Any] = {}
    for logic, chosen in (picks or {}).items():
        kept = {k: v for k, v in (chosen or {}).items() if int(k) < lines.get(logic, 0)}
        if kept:
            out[logic] = kept
    return out


def reapply(project_root: str | Path, folder: str) -> dict[str, Any]:
    """재적용 — 원본(층 · 마스터)을 다시 가져옴 · 입력값 · 고른자재는 같은 키면 살림.

    원본과 달라 덮인 파일(도면 · 면적 · 재료 · 자재조합 · 입력연결)은 `덮임` 에 알림."""
    target = folder_of(project_root, folder)
    manifest = read_json(target / MANIFEST_NAME) or {}
    values = read_json(target / VALUES_FILE)
    picks = read_json(target / PICKS_FILE)
    before = {name: read_json(target / file) for name, file in KINDS.items()}
    files = build_snapshot(manifest["원본"])
    labor = files.get(LABOR_FILE) or {}
    files[VALUES_FILE] = _kept_values(values, labor)
    files[PICKS_FILE] = _kept_picks(picks, labor)
    for old in target.glob("*.json"):  # 원본에서 빠진 파일(도면 없앰 등)은 지움
        if old.name != MANIFEST_NAME and old.name not in files:
            old.unlink()
    for name, content in files.items():
        atomic_write_json(target / name, content)
    covered = [
        name
        for name, file in KINDS.items()
        if file not in LOCAL_FILES and before[name] != files.get(file)
    ]
    source = version_of(source_files(files))
    stamp = now_text()
    fresh = {**manifest, "판": source, "원본판": source, "가져온날": stamp}
    atomic_write_json(target / MANIFEST_NAME, fresh)
    return {"폴더": folder, "가져온날": stamp, "원본판": source, "덮임": covered}


refresh = reapply  # 25-1b 옛 이름(새로 가져오기)


def prefetch(project_root: str | Path, owner_id: str, item: dict[str, Any]) -> dict[str, Any]:
    """구조물 복사본을 미리 가져옴(아직 [저장] 전) — 같은 항목이면 그대로 · 다르면 새로.

    [저장] 뒤에도 그 구조물이 이 항목을 안 쓰면 맞추기가 지움(고친 흔적이 있으면 `_지난/`)."""
    folder = folder_name(owner_id)
    if not FOLDER_PATTERN.match(folder):
        raise ValueError(f"구조물 id 모양이 아님 「{owner_id}」")
    target = _migrated(project_root) / folder
    manifest = read_json(target / MANIFEST_NAME)
    made = not (isinstance(manifest, dict) and same_item(manifest, item))
    if made:
        if target.is_dir():
            retire(target)
        manifest = write_snapshot(target, item, owner_id, {"미리가져옴": now_text()})
    return {"폴더": folder, "만듦": made, "가져온날": manifest.get("가져온날")}


def save_as(
    project_root: str | Path, folder: str, company: int, user: int, path: list[str], name: str
) -> dict[str, Any]:
    """[개인에 저장] — 복사본 내용으로 개인 층에 새 항목(단가 · 입력값 · 고른자재는 안 가져감)."""
    target = folder_of(project_root, folder)
    manifest = read_json(target / MANIFEST_NAME) or {}
    made = personal.create_item(
        personal.personal_folder(company, user),
        path,
        name,
        None,
        copy=target,
        origin=manifest.get("원본"),
    )
    return {**made, "summary_item": personal.personal_item(company, user, made)}


def new_item(
    project_root: str | Path,
    owner_id: str,
    company: int,
    user: int,
    path: list[str],
    name: str,
    unit: str | None,
) -> dict[str, Any]:
    """프로젝트에서 새 항목 — 개인 층에 빈 항목을 만들고 그 구조물 복사본으로.

    돌려준 `summary_item` 을 화면이 그 구조물에 지정해 [저장] 해야 복사본이 남음(안 하면 맞추기가
    다른 항목 · 지정 없음으로 보고 지움)."""
    made = personal.create_item(personal.personal_folder(company, user), path, name, unit)
    item = personal.personal_item(company, user, made)
    got = prefetch(project_root, owner_id, item)
    return {**made, "폴더": got["폴더"], "summary_item": item}


def calc(project_root: str | Path, folder: str) -> dict[str, Any]:
    """복사본(입력값 · 고른자재 · 단가 · 일위대가)만으로 계산 — 마스터 현재값을 안 읽음."""
    target = folder_of(project_root, folder)
    labor = read_json(target / LABOR_FILE)
    if not isinstance(labor, dict):
        raise LookupError(f"일위대가 조합이 없는 항목 「{folder}」 — 계산할 것이 없음")
    return prices.calc(
        labor,
        read_json(target / PRICE_FILE) or {},
        read_json(target / VALUES_FILE) or {},
        read_json(target / PICKS_FILE) or {},
    )