Aislo/D03_PlanAnalysis/D03_PlanAnalysis_Store.py

"""D 예정노선 설계 내용 한 파일 — `<프로젝트>/D03_PlanAnalysis/plan_network.json`
(요구 452 · 453 · PLAN 590-1).

- 꼴 = `{version: 2, epsg, points: [{x, y}], lines, marks,
  pieces: [{id, line, from_m, to_m, year, set}]}` — 좌표계 = 머리 한 곳 · 연도 = 조각의 `year` 뿐 ·
  세트 = 조각의 `set`(열쇠 「연도-번호」 · PLAN 638-1 · 판 1 = `set` 없음 → 읽을 때 번호) ·
  길이 · 측점 · 조각 좌표 · 세트 덩이 · 연도 지문은 읽을 때 `plan_state` 가 셈(저장 안 함).
- DB 에 남는 것 = 등록 칸 · 연도 상태(`plan_route_years` 의 year · fingerprint · confirmed_at) ·
  업체 줄(`plan_route_assignees.piece_id` = 세트 열쇠(옛 줄 = 조각 id → `find_unit` 이 그 세트로) ·
  업체 · 메일 시각 · 연결 — 연도는 이 파일에서).
- 파일이 없으면 빈 설계(새 프로젝트) · 파일은 첫 쓰기 때 생김(옛 DB 칸 옮김 = 590-3 에서 끝남).
- 쓰기 = `project_lock(project_id)` 안에서 `save_plan`(같은 폴더 임시 파일 → 바꿔치기 · 반쯤 쓴
  파일이 안 보임) · 읽기는 잠금 없음.
- 확정 = 연도 상태 줄 지문이 파일 지문과 같을 때만(파일만 바뀌고 DB 가 못 따라간 틈 =
  풀린 것으로 봄).
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any
from uuid import UUID

import aiomysql
from fastapi import HTTPException

from common_util.common_util_json import atomic_write_json
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool
from D03_PlanAnalysis import D03_PlanAnalysis_Network as network
from D04_PlanConfirm.D04_PlanConfirm_Repository import line_names

FILE = Path("D03_PlanAnalysis") / "plan_network.json"
VERSION = 2  # 2 = 조각 `set`(옛 서버가 덮어 세트를 지우지 않게 · 638-1)
KEYS = ("epsg", "points", "lines", "marks", "pieces")


def empty_plan(epsg: int | None = None) -> dict[str, Any]:
    return {"version": VERSION, "epsg": epsg, "points": [], "lines": [], "marks": [], "pieces": []}


def plan_path(root: str | Path) -> Path:
    return Path(root) / FILE


def _clean(plan: dict[str, Any]) -> dict[str, Any]:
    out = empty_plan()
    out.update({key: plan[key] for key in KEYS if plan.get(key) is not None})
    out["epsg"] = int(out["epsg"]) if out["epsg"] is not None else None
    return out


def read_plan(root: str | Path) -> dict[str, Any] | None:
    """파일 — 없으면 None · 이 코드보다 새 판이면 거절(덮어써 잃지 않게)."""
    path = plan_path(root)
    if not path.is_file():
        return None
    plan = json.loads(path.read_text("utf-8"))
    if int(plan.get("version") or 0) > VERSION:
        raise HTTPException(status_code=409, detail="새 판 노선 파일입니다 — 서버를 새로 받으세요.")
    return _clean(plan)


def save_plan(root: str | Path, plan: dict[str, Any]) -> None:
    atomic_write_json(plan_path(root), _clean(plan))


async def _project(cursor: Any, project_id: UUID | str, plan_only: bool = True) -> dict[str, Any]:
    """프로젝트 줄 — `plan_only` = 예정노선 아니면 400(지도 겹 길은 종류 안 가림 · 624-1)."""
    await cursor.execute(
        "SELECT project_kind, project_year, storage_path FROM projects"
        " WHERE id = %s AND deleted_at IS NULL",
        (str(project_id),),
    )
    project = await cursor.fetchone()
    if not project:
        raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
    if plan_only and (project["project_kind"] != "PLAN" or project["project_year"] is None):
        raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.")
    if not project["storage_path"]:
        raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
    return project


async def load_plan(project_id: UUID | str) -> tuple[Path, dict[str, Any]]:
    """(프로젝트 실경로, 설계 내용) — 파일 · 없으면 빈 설계(쓰지 않음)."""
    async with (
        get_db_pool().acquire() as connection,
        connection.cursor(aiomysql.DictCursor) as cursor,
    ):
        project = await _project(cursor, project_id)
    root = Path(resolve_stored_project_path(project["storage_path"]))
    return root, read_plan(root) or empty_plan()


async def project_row(project_id: UUID | str) -> dict[str, Any]:
    """{project_kind, project_year, storage_path} — 종류 안 가림 · 없음 404."""
    async with (
        get_db_pool().acquire() as connection,
        connection.cursor(aiomysql.DictCursor) as cursor,
    ):
        return await _project(cursor, project_id, plan_only=False)


def plan_parts(plan: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """(조각, 세트 단위) — 틀리면 `NetworkError`.

    조각 = {id · line · line_name · year · set · set_key · epsg · from_m · to_m · length_m ·
    planned · stations} · 연도 · 선 차례 · 선 위 거리 차례.
    세트 단위 = {id(「2027-1」) · set · year · line · line_name(첫 조각) · epsg · from_m · to_m ·
    length_m · planned(이은 한 줄 · 덩이 2 이상 · 갈래면 빈 것) · parts · branch ·
    pieces(조각 id)} —
    여러 조각 세트의 to_m = from_m + 길이."""
    if not plan["lines"]:
        return [], []
    derived = network.derive(plan)
    pieces = derived["pieces"]
    names = dict(zip((line["id"] for line in plan["lines"]), line_names(plan["lines"])))
    for piece in pieces:
        piece.update(line_name=names.get(piece["line"], "갈래"), epsg=plan["epsg"])
    pieces.sort(key=lambda p: p["year"])  # 같은 연도 안 = derive 차례(선 · 거리) 그대로
    by_id = {p["id"]: p for p in pieces}
    units = []
    for item in derived["sets"]:
        parts = [by_id[i] for i in item["pieces"]]
        first = parts[0]
        whole = item["parts"] == 1 and not item["branch"]
        units.append(
            {
                "id": item["key"],
                "set": item["no"],
                "year": item["year"],
                "line": first["line"],
                "line_name": first["line_name"],
                "epsg": plan["epsg"],
                "from_m": first["from_m"],
                "to_m": first["to_m"] if len(parts) == 1 else first["from_m"] + item["length_m"],
                "length_m": item["length_m"],
                "planned": (network.join_set(parts) or []) if whole else [],
                "parts": item["parts"],
                "branch": item["branch"],
                "pieces": item["pieces"],
            }
        )
    return pieces, units


def plan_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
    """조각만(`plan_parts` 앞 값)."""
    return plan_parts(plan)[0]


def find_unit(units: list[dict[str, Any]], key: str | None, field: str = "id") -> Any:
    """업체 줄 열쇠 → 세트 단위 — 세트 열쇠 · 옛 조각 id(그 조각이 든 세트) · 없으면 None."""
    return next((u for u in units if u[field] == key), None) or next(
        (u for u in units if key in u["pieces"]), None
    )


async def year_status(project_id: UUID | str) -> dict[int, dict[str, Any]]:
    """연도 상태 줄 — {연도: {fingerprint, confirmed_at}}."""
    async with (
        get_db_pool().acquire() as connection,
        connection.cursor(aiomysql.DictCursor) as cursor,
    ):
        await cursor.execute(
            "SELECT year, fingerprint, confirmed_at FROM plan_route_years WHERE project_id = %s",
            (str(project_id),),
        )
        return {int(r["year"]): r for r in await cursor.fetchall()}


async def plan_state(project_id: UUID | str) -> dict[str, Any]:
    """{root, plan, pieces, units, years} — `units` = 세트 단위(`plan_parts`) ·
    `years` = `year_rows` + `confirmed_at`(지문이 같을 때만) · 단위 · 조각마다 `confirmed_at` ·
    노선이 틀리면 409."""
    root, plan = await load_plan(project_id)
    try:
        pieces, units = plan_parts(plan)
    except network.NetworkError as error:
        raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
    years = network.year_rows(pieces)
    status = await year_status(project_id)
    for year, row in years.items():
        saved = status.get(year)
        same = saved is not None and saved["fingerprint"] == row["fingerprint"]
        row["confirmed_at"] = saved["confirmed_at"] if same else None
    for unit in (*pieces, *units):
        unit["confirmed_at"] = years[unit["year"]]["confirmed_at"]
    return {"root": root, "plan": plan, "pieces": pieces, "units": units, "years": years}