Aislo/D03_PlanAnalysis/D03_PlanAnalysis_Store.py

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

- 꼴 = `{version, epsg, points: [{x, y}], lines, marks, pieces: [{id, line, from_m, to_m, year}]}` —
  좌표계 = 머리 한 곳 · 연도 = 조각의 `year` 뿐(빈 연도 줄 없음) · 길이 · 측점 · 조각 좌표 ·
  연도 지문은 읽을 때 `plan_state` 가 셈(저장 안 함).
- DB 에 남는 것 = 등록 칸 · 연도 상태(`plan_route_years` 의 year · fingerprint · confirmed_at) ·
  업체 줄(`plan_route_assignees` 조각 번호 · 업체 · 메일 시각 · 연결 — 연도는 이 파일에서).
- 파일이 없으면 빈 설계(새 프로젝트) · 파일은 첫 쓰기 때 생김(옛 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 = 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_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
    """조각마다 {id · line · line_name · year · epsg · from_m · to_m · length_m · planned ·
    stations}
    — 연도 · 선 차례 · 선 위 거리 차례 · 틀리면 `NetworkError`."""
    pieces = network.derive(plan)["pieces"] if plan["lines"] else []
    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 차례(선 · 거리) 그대로
    return pieces


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, units, years} — `years` = `year_rows` + `confirmed_at`(지문이 같을 때만) ·
    단위마다 `confirmed_at` · 노선이 틀리면 409."""
    root, plan = await load_plan(project_id)
    try:
        units = plan_units(plan)
    except network.NetworkError as error:
        raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
    years = network.year_rows(units)
    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 units:
        unit["confirmed_at"] = years[unit["year"]]["confirmed_at"]
    return {"root": root, "plan": plan, "units": units, "years": years}