Aislo/B08_Quantity/B08_Quantity_Router_Milestone.py

"""B08 거리이정표 길(PLAN 244-1) — 공사지 · 기준점 · 경로(업체) · 표 · 경로 그림. DB 표 없음.

    GET    /api/projects/{id}/quantity/milestone                 {data, site, table, images}
    PUT    /api/projects/{id}/quantity/milestone                 [저장] — 입력만 · 표는 서버가 셈
    POST   /api/projects/{id}/quantity/milestone/image?route=    경로 그림 올리기(하나 · 교체)
    GET    /api/projects/{id}/quantity/milestone/image?route=    그림 내려받기
    DELETE /api/projects/{id}/quantity/milestone/image?route=    그림 지우기

정본 = 프로젝트 storage `B08_Quantity/milestone/`(`milestone.json` + `<경로 id>.<확장자>` 그림 ·
폴더가 목록 · B09 견적서 파일 길 꼴). 초기값 층 없음(자동설계가 낳는 값 아님).
그림은 올리는 즉시 서버에([저장]과 따로) · [저장] 때 없는 경로의 그림은 지움.
공사지 = 시점(노선 첫 점 · 측점 0) · 중점 기본 = 노선 연장 ÷ 2 — 노선은 설계 노선 정본 차례
(수정본 → 초기 폴리라인 → 예상노선 → 초기값 · `load_design_route` 와 같은 차례).
표 합 = `B08_Quantity_Milestone_Calc`(파이썬 · TS 짝).
"""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
from typing import Any, Literal
from uuid import UUID

from fastapi import APIRouter, File, HTTPException, Query, UploadFile
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field, ValidationError, model_validator

from B03_FileInput.B03_FileInput_Engine import save_upload_stream
from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone, r3
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from common_util.common_util_initial_snapshot import design_route_csv_path
from common_util.common_util_json import atomic_write_json
from common_util.common_util_route_geometry import (
    expected_route_csv_path,
    interpolate_vertex,
    planned_route_initial_path,
    planned_route_working_path,
    read_planned_route_csv,
)

router = APIRouter(prefix="/api/projects", tags=["B08 Milestone"])
PATH = "/{project_id}/quantity/milestone"
FOLDER = Path("B08_Quantity") / "milestone"
DATA_FILE = "milestone.json"
IMAGE_EXT = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp")
ROUTE_ID = r"^[A-Za-z0-9_-]{1,40}$"

Road = Literal["2차포장", "1차포장", "비포장"]
Dir = Literal["N", "NE", "E", "SE", "S", "SW", "W", "NW"]


class Leg(BaseModel):
    road: Road = "2차포장"
    km: float | None = Field(default=None, ge=0, le=10000)


class Row(BaseModel):
    item: str = Field(default="", max_length=100)
    dest: Literal["start", "mid"] = "start"
    note: str = Field(default="", max_length=200)


class Place(BaseModel):
    name: str = Field(default="", max_length=100)
    address: str = Field(default="", max_length=200)
    dir: Dir | None = None
    lon: float | None = Field(default=None, ge=-180, le=180)
    lat: float | None = Field(default=None, ge=-90, le=90)
    legs: list[Leg] = Field(default_factory=list, max_length=20)


class Route(Place):
    id: str = Field(pattern=ROUTE_ID)
    via_base: bool = True
    rows: list[Row] = Field(default_factory=list, max_length=50)


class Base(Place):
    enabled: bool = False


class Site(BaseModel):
    mid_station_m: float | None = Field(default=None, ge=0, le=1_000_000)
    mid_road: Road = "비포장"


class Milestone(BaseModel):
    site: Site = Field(default_factory=Site)
    base: Base = Field(default_factory=Base)
    routes: list[Route] = Field(default_factory=list, max_length=100)

    @model_validator(mode="after")
    def _unique_ids(self) -> Milestone:
        ids = [route.id for route in self.routes]
        if len(ids) != len(set(ids)):
            raise ValueError("경로 id 가 겹칩니다.")
        return self


def folder(root: str | Path) -> Path:
    return Path(root) / FOLDER


def read_data(root: str | Path) -> dict[str, Any]:
    """저장본 — 없거나 깨졌으면 빈 꼴."""
    path = folder(root) / DATA_FILE
    try:
        return Milestone.model_validate_json(path.read_text(encoding="utf-8")).model_dump()
    except (OSError, ValidationError, json.JSONDecodeError):
        return Milestone().model_dump()


def _route_vertices(root: Path) -> list[Any]:
    for path in (
        planned_route_working_path(root),
        planned_route_initial_path(root),
        expected_route_csv_path(root),
        design_route_csv_path(root),
    ):
        route = read_planned_route_csv(path) if path.is_file() else None
        if route is not None and len(route.vertices) >= 2:
            return route.vertices
    return []


def _lonlat(root: Path, x: float, y: float) -> list[float] | None:
    """작업 좌표계 → 경도 · 위도(다이어그램 방위 · 244-3 지도) — 못 하면 None."""
    try:
        from pyproj import Transformer

        from B04_PreProcess.B04_PreProcess_Engine_Extent import project_epsg_from_prj

        crs = project_epsg_from_prj(root)
        lon, lat = Transformer.from_crs(crs, "EPSG:4326", always_xy=True).transform(x, y)
        return [round(lon, 7), round(lat, 7)]
    except Exception:  # 좌표계 못 읽음 = 좌표 없음(방위는 사용자 8방)
        return None


def site_info(root: str | Path, mid_station_m: float | None) -> dict[str, Any]:
    """노선 연장 · 중점 기본(연장 ÷ 2) · 시점 · 중점 경위도."""
    root = Path(root)
    vertices = _route_vertices(root)
    if not vertices:
        return {"length_m": None, "mid_default_m": None, "start_lonlat": None, "mid_lonlat": None}
    length = vertices[-1].chainage_m
    mid_default = r3(length / 2)
    mx, my, _ = interpolate_vertex(
        vertices, mid_default if mid_station_m is None else mid_station_m
    )
    return {
        "length_m": r3(length),
        "mid_default_m": mid_default,
        "start_lonlat": _lonlat(root, vertices[0].x, vertices[0].y),
        "mid_lonlat": _lonlat(root, mx, my),
    }


def route_images(root: str | Path, route_id: str) -> list[Path]:
    path = folder(root)
    if not path.is_dir():
        return []
    return sorted(
        f for f in path.glob(f"{route_id}.*") if f.is_file() and f.suffix.lower() in IMAGE_EXT
    )


def images(root: str | Path) -> dict[str, str]:
    """경로 id → 그림 파일 이름(폴더가 목록)."""
    path = folder(root)
    if not path.is_dir():
        return {}
    return {
        f.stem: f.name
        for f in sorted(path.iterdir())
        if f.is_file() and f.suffix.lower() in IMAGE_EXT and not f.name.startswith(".")
    }


def payload(root: str | Path, data: dict[str, Any]) -> dict[str, Any]:
    site = site_info(root, data["site"]["mid_station_m"])
    return {
        "data": data,
        "site": site,
        "table": compute_milestone(data, site["mid_default_m"]),
        "images": images(root),
    }


def save_data(root: str | Path, body: Milestone) -> dict[str, Any]:
    data = body.model_dump()
    path = folder(root)
    path.mkdir(parents=True, exist_ok=True)
    atomic_write_json(path / DATA_FILE, data)
    keep = {route["id"] for route in data["routes"]}
    for stem, name in images(root).items():
        if stem not in keep:
            (path / name).unlink(missing_ok=True)
    return payload(root, data)


def _route_id(route: str) -> str:
    try:
        return Route(id=route).id
    except ValidationError as error:
        raise HTTPException(status_code=400, detail="경로 id 가 바르지 않습니다.") from error


@router.get(PATH)
async def milestone_read(project_id: UUID) -> dict[str, Any]:
    root = await _project_root(project_id)
    return await asyncio.to_thread(lambda: payload(root, read_data(root)))


@router.put(PATH)
async def milestone_save(project_id: UUID, body: Milestone) -> dict[str, Any]:
    root = await _project_root(project_id)
    return await asyncio.to_thread(save_data, root, body)


@router.post(PATH + "/image")
async def milestone_image_upload(
    project_id: UUID, route: str = Query(...), file: UploadFile = File(...)
) -> dict[str, Any]:
    route_id = _route_id(route)
    ext = Path(file.filename or "").suffix.lower()
    if ext not in IMAGE_EXT:
        raise HTTPException(status_code=400, detail="그림 파일(png · jpg · gif · webp · bmp)만.")
    root = await _project_root(project_id)
    path = folder(root)
    path.mkdir(parents=True, exist_ok=True)
    try:
        await save_upload_stream(file, path / f".{route_id}{ext}.new")
    except ValueError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error
    for old in route_images(root, route_id):
        old.unlink(missing_ok=True)
    (path / f".{route_id}{ext}.new").replace(path / f"{route_id}{ext}")
    return {"name": f"{route_id}{ext}"}


@router.get(PATH + "/image")
async def milestone_image(project_id: UUID, route: str = Query(...)):
    route_id = _route_id(route)
    root = await _project_root(project_id)
    found = route_images(root, route_id)
    if not found:
        raise HTTPException(status_code=404, detail="경로 그림이 없습니다.")
    return FileResponse(found[0])


@router.delete(PATH + "/image")
async def milestone_image_delete(project_id: UUID, route: str = Query(...)) -> dict[str, Any]:
    route_id = _route_id(route)
    root = await _project_root(project_id)
    for old in route_images(root, route_id):
        old.unlink(missing_ok=True)
    return {"name": None}