Aislo/M02_MasterTemplete/M02_Item_Snapshot.py

"""구조물 집계표 항목 스냅샷 — 프로젝트 `templates/items/{폴더}/` (PLAN 21-1 · 21-3 · 25-2b).

[저장](`PUT /route/structures` · `PUT /drainage/pipe-points`) 뒤 서버가 맞춤:
    - 구조물 · 관 지점이 가리키는 항목(`summary_item`)마다 원본 층에서 내용을 복사
      (`M02_Item_Source.build_source` — 집계 · 도면 · 면적 · 재료 · 자재조합 · 일위대가 · 입력연결 ·
      단가) + 프로젝트 값 `입력값.json`(처음엔 빈 값 · 자동값은 안 적음) · `고른자재.json` ·
      `manifest.json`
    - 이미 있는 폴더는 내용을 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌어도 프로젝트
      값은 지정한 때 그대로 · `manifest` 의 쓰는 구조물 목록만 새로([새로 가져오기]는 사용자가)
    - 아무 구조물도 안 쓰는 폴더는 지움(지정 바뀜 · 해제 · 구조물 삭제) — 단 프로젝트에서 고친
      흔적(`manifest.고침`)이 있으면 남김
원본 층 자리 = `M02_Template_Layers` 경로(시스템 층은 지금 파일) · 일위대가 · 단가는 M01 시스템 것.
"""

from __future__ import annotations

import logging
import shutil
from datetime import datetime
from pathlib import Path
from typing import Any

from B05_Profile.B05_Profile_Structures_Repository import load_structures
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
from common_util.common_util_json import atomic_write_json
from common_util.common_util_summary_item import snapshot_folder_name
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_Item_Source import (
    LINK_FILE,
    PRICE_FILE,
    build_source,
    read_json,
    version_of,
)

logger = logging.getLogger(__name__)

ITEMS_DIRNAME = "items"
MANIFEST_NAME = "manifest.json"
VALUES_FILE = "입력값.json"  # {로직키: {입력이름: {값, 확인}}}
PICKS_FILE = "고른자재.json"  # {로직키: {호표 차례: MT키}}


def items_dir(project_root: str | Path) -> Path:
    return layers.project_dir(project_root) / ITEMS_DIRNAME


def now_text() -> str:
    return datetime.now().isoformat(timespec="seconds")


def build_snapshot(item: dict[str, Any]) -> dict[str, Any]:
    """원본 내용 + 빈 프로젝트 값 — `{파일 이름: 내용}` · SS 줄이 없으면 LookupError."""
    return {**build_source(item), VALUES_FILE: {}, PICKS_FILE: {}}


def write_snapshot(target: Path, item: dict[str, Any], users: list[str], extra: dict) -> dict:
    """폴더에 새로 씀 — manifest 는 마지막(있으면 다 쓴 폴더 · 끊기면 다음 저장이 다시 씀)."""
    files = build_snapshot(item)
    target.mkdir(parents=True, exist_ok=True)
    for name, content in files.items():
        atomic_write_json(target / name, content)
    source = {k: v for k, v in files.items() if k not in (VALUES_FILE, PICKS_FILE)}
    stamp = now_text()
    manifest = {
        "원본": item,
        "판": version_of(source),
        "원본판": version_of(source),
        "적용일": stamp,
        "가져온날": stamp,
        "구조물": users,
        **extra,
    }
    atomic_write_json(target / MANIFEST_NAME, manifest)
    return manifest


def assignments(project_root: str | Path) -> dict[str, tuple[dict[str, Any], list[str]]]:
    """`{폴더 이름: (항목, 쓰는 구조물 id)}` — 구조물 정본 + 관 지점 정본(id = `pipe@측점`)."""
    found: dict[str, tuple[dict[str, Any], list[str]]] = {}

    def add(item: dict[str, Any] | None, owner_id: str) -> None:
        if item:
            found.setdefault(snapshot_folder_name(item), (item, []))[1].append(owner_id)

    _, structures = load_structures(str(project_root))
    for index, structure in enumerate(structures):
        add(structure.summary_item, structure.structure_id or f"#{index}")
    for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))):
        add(point.summary_item, f"pipe@{round(float(point.chainage_m), 3)}")
    return found


def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
    """지정 항목 스냅샷을 맞춤 — `{만듦, 지움, 못만듦}` 폴더 이름."""
    base = items_dir(project_root)
    wanted = assignments(project_root)
    made: list[str] = []
    failed: list[str] = []
    for folder, (item, users) in sorted(wanted.items()):
        target = base / folder
        manifest = read_json(target / MANIFEST_NAME)
        if isinstance(manifest, dict):
            if manifest.get("구조물") != users:
                atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": users})
            continue
        try:
            write_snapshot(target, item, users, {})
        except (LookupError, ValueError) as error:
            logger.warning("집계 항목 스냅샷을 못 만듦 (%s): %s", folder, error)
            failed.append(folder)
            continue
        made.append(folder)
    removed: list[str] = []
    if base.is_dir():
        for child in sorted(base.iterdir()):
            if not child.is_dir() or child.name in wanted:
                continue
            manifest = read_json(child / MANIFEST_NAME)
            if isinstance(manifest, dict) and manifest.get("고침"):
                if manifest.get("구조물"):  # 고친 복사본은 남기되 쓰는 구조물은 비움
                    atomic_write_json(child / MANIFEST_NAME, {**manifest, "구조물": []})
                continue
            shutil.rmtree(child)
            removed.append(child.name)
    return {"만듦": made, "지움": removed, "못만듦": failed}


def safe_sync(project_root: str | Path) -> None:
    """저장 뒤 부름 — 실패해도 저장은 그대로(스냅샷은 다음 저장이 다시 맞춤)."""
    try:
        result = sync_snapshots(project_root)
        if any(result.values()):
            logger.info("집계 항목 스냅샷: %s", result)
    except Exception:
        logger.exception("집계 항목 스냅샷 맞춤 실패 (저장은 유지): %s", project_root)


def complete_old(folder: Path) -> bool:
    """옛 스냅샷(21-3 · 새 파일 없음)을 채움 — 입력연결 · 단가는 지금 원본에서 · 값은 빈 것.

    옛 내용 파일(집계 · 도면 …)은 그대로 둠. 채웠으면 True(manifest 에 `옮겨받음` 날).
    """
    manifest = read_json(folder / MANIFEST_NAME)
    if not isinstance(manifest, dict):
        return False
    lacking = [
        name
        for name in (LINK_FILE, PRICE_FILE, VALUES_FILE, PICKS_FILE)
        if not (folder / name).is_file()
    ]
    if not lacking and "원본판" in manifest:
        return False
    source = (
        build_source(manifest["원본"]) if (LINK_FILE in lacking or PRICE_FILE in lacking) else {}
    )
    for name in lacking:
        atomic_write_json(folder / name, source.get(name, {}))
    kept = {
        name: read_json(folder / name)
        for name in sorted(p.name for p in folder.glob("*.json"))
        if name not in (MANIFEST_NAME, VALUES_FILE, PICKS_FILE)
    }
    atomic_write_json(
        folder / MANIFEST_NAME,
        {
            **manifest,
            "가져온날": manifest.get("가져온날") or manifest.get("적용일"),
            "원본판": manifest.get("원본판") or version_of(kept),
            "옮겨받음": now_text(),
        },
    )
    return True


def read_snapshot(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any] | None:
    """프로젝트 스냅샷의 `집계.json`(`{줄, 열}`) — 없으면 None."""
    data = read_json(items_dir(project_root) / snapshot_folder_name(item) / "집계.json")
    return data if isinstance(data, dict) else None