Aislo/D03_PlanAnalysis/D03_PlanAnalysis_Router.py

"""D 예정노선 2단계 지형 API (요구 345 · 389 · PLAN 530-1 · 561-2).

- `GET /api/projects/{id}/plan/terrain/extent` — 받기 전 범위(상자 · 도엽 · 칸 수 · 받은 범위)
- `POST …/plan/terrain` — 사업 전체 지형 한 벌 받기(뒤에서 · 누를 때만) · 끝나면 `surface_models` 행
- `GET …/plan/progress` — 일의 진행률
- `GET …/plan/layers/sheet?key=` · `…/cadastral` · `…/raster?kind=slope|hillshade|landslide` ·
  `…/zones` · `…/soils` · `…/bands` · `…/lines?kind=` · `…/zero-order` ·
  `…/forest-roads`(GET · POST 올리기) · `…/landslide-zones` — 지도 겹
  (PLAN 569-1 · 581-1 · 581-2 · 584-1 · 584-3)

범위 = 전체 시점 · 종점(D02 정본 `load_base`) + 그은 점(`extra`) 상자 + 여유.
노선 정본(손 그은 선)은
`D03_PlanAnalysis_Network`(PLAN 562-2 · 옛 추천 엔진 · 꺾임점 R 걷음).
"""

from __future__ import annotations

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

from fastapi import APIRouter, File, HTTPException, Request, Response, UploadFile
from pydantic import BaseModel

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from common_util.common_util_http_cache import cached_file_response
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool
from D02_PlanBase import D02_PlanBase_Router as base
from D03_PlanAnalysis import D03_PlanAnalysis_ForestRoad as forest_road
from D03_PlanAnalysis import D03_PlanAnalysis_Landslide as landslide
from D03_PlanAnalysis import D03_PlanAnalysis_LandslideZone as landslide_zone
from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers
from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero_order

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["D03 Plan Analysis"])

# 진행 — (프로젝트, 일) → {running, percent, message, error} · 프로세스 메모리
_JOBS: dict[tuple[str, str], dict[str, Any]] = {}
_TASKS: set[asyncio.Task] = set()
KINDS = ("terrain", "analyze")


class TerrainBody(BaseModel):
    """지형 받기 — 여유(m · 빈 값 = 처음 값) · 더 덮을 점(그은 꺾임점 · 작업 좌표계 m)."""

    margin_m: float | None = None
    extra: list[tuple[float, float]] = []


async def _plan_ends(project_id: UUID) -> tuple[Path, int, list[tuple[float, float]]]:
    """(실경로, 좌표계, [전체 시점들 · 종점들]) — D02 정본(빈 칸이면 옛 연도 행 접음)."""
    plan = await base.load_base(project_id)
    if not plan["starts"] or not plan["ends"]:
        raise HTTPException(
            status_code=400, detail="전체 시점 · 종점이 없습니다 — 1단계에서 지정하세요."
        )
    async with get_db_pool().acquire() as connection:
        stored = await get_project_storage_relative_path(connection, project_id)
    if not stored:
        raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
    ends = [(float(p["x"]), float(p["y"])) for p in (*plan["starts"], *plan["ends"])]
    return Path(resolve_stored_project_path(stored)), int(plan["epsg"]), ends


def _query_xy(extra: str | None) -> list[tuple[float, float]]:
    if not extra:
        return []
    try:
        return [(float(x), float(y)) for x, y in json.loads(extra)]
    except (ValueError, TypeError) as error:
        raise HTTPException(status_code=400, detail="extra 꼴이 다릅니다.") from error


def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
    key = (str(project_id), kind)
    if (_JOBS.get(key) or {}).get("running"):
        raise HTTPException(status_code=409, detail="이미 돌고 있습니다.")
    job = {"running": True, "percent": 0, "message": "시작", "error": None}
    _JOBS[key] = job

    def report(percent: int, message: str) -> None:
        job.update(percent=int(percent), message=message)

    async def run() -> None:
        try:
            await work(report)
            job.update(percent=100)
        except Exception as error:  # noqa: BLE001 — 진행률에 까닭을 그대로 올림
            logger.exception("D %s 실패: project_id=%s", kind, project_id)
            job["error"] = str(getattr(error, "detail", None) or error)
        finally:
            job["running"] = False

    task = asyncio.create_task(run(), name=f"D {kind} {project_id}")
    _TASKS.add(task)
    task.add_done_callback(_TASKS.discard)
    return {"status": "accepted", **job}


async def _save_models(project_id: UUID, models: list[dict[str, Any]], epsg: int) -> None:
    """지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로)."""
    from B04_PreProcess.B04_PreProcess_Repository import _existing_models, _upsert_surface_model

    async with get_db_pool().acquire() as connection:
        await connection.begin()
        try:
            existing = await _existing_models(connection, project_id)
            for model in models:
                await _upsert_surface_model(
                    connection,
                    existing,
                    project_id=project_id,
                    model=model,
                    source_file_id=None,
                    processed_cloud_id=None,
                    crs_epsg=epsg,
                )
            await connection.commit()
        except Exception:
            await connection.rollback()
            raise


@router.get("/{project_id}/plan/terrain/extent", response_model=None)
async def plan_terrain_extent(
    project_id: UUID, margin_m: float | None = None, extra: str | None = None
) -> dict[str, Any]:
    """받기 전 범위 — 전체 시점 · 종점(+ `extra` 그은 점 `[[x, y]]`) 상자 + 여유 ∪ 받은 범위."""
    root, epsg, ends = await _plan_ends(project_id)
    view = await asyncio.to_thread(
        terrain.plan_extent, root, [*ends, *_query_xy(extra)], epsg, margin_m
    )
    return {"status": "success", **view}


@router.post("/{project_id}/plan/terrain", response_model=None)
async def plan_terrain(project_id: UUID, body: TerrainBody | None = None) -> dict[str, Any]:
    """사업 전체 수치지형도 · 위성 · DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
    body = body or TerrainBody()
    root, epsg, ends = await _plan_ends(project_id)
    points = [*ends, *body.extra]

    async def work(report: Any) -> None:
        got = await asyncio.to_thread(
            terrain.ensure_plan_terrain, root, points, epsg, body.margin_m, report
        )
        await _save_models(project_id, got["models"], epsg)
        report(100, f"지형 받기 끝 — 새 도엽 {len(got['sheets']['downloaded'])}매")

    return _start_job(project_id, "terrain", work)


@router.get("/{project_id}/plan/progress", response_model=None)
async def plan_progress(project_id: UUID) -> dict[str, Any]:
    idle = {"running": False, "percent": 0, "message": None, "error": None}
    return {
        "status": "success",
        **{kind: dict(_JOBS.get((str(project_id), kind)) or idle) for kind in KINDS},
    }


async def _layer_root(project_id: UUID) -> tuple[Path, int, terrain.Box]:
    """(실경로, 좌표계, 받은 범위) — 지형 받기 전이면 404."""
    root, epsg, _ = await _plan_ends(project_id)
    box = terrain.received_box(root)
    if box is None:
        raise HTTPException(status_code=404, detail="지형 받기 뒤 보입니다.")
    return root, epsg, box


@router.get("/{project_id}/plan/layers/sheet", response_model=None)
async def plan_layer_sheet(project_id: UUID, key: str, request: Request) -> Response:
    """도엽 겹 GeoJSON(하천중심선 · 도로중심선 …) — 없으면 지난 병합 도엽에서 풀어 냄."""
    from B04_PreProcess.B04_PreProcess_Router_GIS import _display_copy

    root, _, _ = await _layer_root(project_id)
    try:
        path = await asyncio.to_thread(layers.ensure_sheet_layer, root, key)
    except KeyError as error:
        raise HTTPException(status_code=400, detail=f"없는 도엽 겹: {key}") from error
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    display = await asyncio.to_thread(_display_copy, path)
    return cached_file_response(request, display, "application/geo+json")


@router.get("/{project_id}/plan/layers/cadastral", response_model=None)
async def plan_layer_cadastral(project_id: UUID, request: Request) -> Response:
    """지적도 GeoJSON(브이월드 LX맵 · 참고용) — 받은 범위를 덮지 않으면 지금 받음."""
    root, epsg, box = await _layer_root(project_id)
    processed = root / terrain.STAGE / "processed"
    try:
        got = await asyncio.to_thread(layers.ensure_cadastral, processed, box, epsg)
    except RuntimeError as error:
        raise HTTPException(status_code=502, detail=str(error)) from error
    return cached_file_response(request, got["path"], "application/geo+json")


@router.get("/{project_id}/plan/layers/raster", response_model=None)
async def plan_layer_raster(project_id: UUID, kind: str = "slope") -> dict[str, Any]:
    """급경사 단계 그림(15° · 35° · 40°) · 음영기복(`hillshade`) — 격자 지문 같으면 저장본 ·
    산사태위험지도(`landslide` · 행안부 WMS) — 받은 범위를 덮으면 저장본."""
    if kind == "landslide":
        root, epsg, box = await _layer_root(project_id)
        processed = root / terrain.STAGE / "processed"
        try:
            value = await asyncio.to_thread(landslide.ensure_landslide, processed, box, epsg)
        except landslide.MissingKeyError as error:
            raise HTTPException(status_code=503, detail=str(error)) from error
        except RuntimeError as error:
            raise HTTPException(status_code=502, detail=str(error)) from error
        return {"status": "success", "kind": kind, **value}
    if kind not in layers.RASTER_KINDS:
        raise HTTPException(status_code=400, detail=f"없는 그림 겹: {kind}")
    root, epsg, _ = await _layer_root(project_id)
    try:
        value = await asyncio.to_thread(
            layers.slope_layer, root, terrain.models_dir_of(root), epsg, kind
        )
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    return {"status": "success", "kind": kind, **value}


@router.get("/{project_id}/plan/layers/zones", response_model=None)
async def plan_layer_zones(project_id: UUID, request: Request) -> Response:
    """보호 · 규제 구역 GeoJSON(여덟 종 · `summary` = 종류별 개수 · 넓이) — 범위 밖이면 받음."""
    root, epsg, box = await _layer_root(project_id)
    processed = root / terrain.STAGE / "processed"
    try:
        got = await asyncio.to_thread(layers.ensure_zones, processed, box, epsg)
    except RuntimeError as error:
        raise HTTPException(status_code=502, detail=str(error)) from error
    return cached_file_response(request, got["path"], "application/geo+json")


@router.get("/{project_id}/plan/layers/soils", response_model=None)
async def plan_layer_soils(project_id: UUID, request: Request) -> Response:
    """산림입지도 GeoJSON(토양형 · `summary` = 토양형별 개수 · 넓이) — 범위 밖이면 받음."""
    root, epsg, box = await _layer_root(project_id)
    processed = root / terrain.STAGE / "processed"
    try:
        got = await asyncio.to_thread(layers.ensure_soils, processed, box, epsg)
    except RuntimeError as error:
        raise HTTPException(status_code=502, detail=str(error)) from error
    return cached_file_response(request, got["path"], "application/geo+json")


@router.get("/{project_id}/plan/layers/bands", response_model=None)
async def plan_layer_bands(project_id: UUID) -> dict[str, Any]:
    """계류 100 m · 도로 300 m 띠(넓이 · 받은 범위 비율) — 두 도엽 겹 지문 같으면 저장본."""
    root, epsg, box = await _layer_root(project_id)
    try:
        value = await asyncio.to_thread(layers.band_layer, root, epsg, box)
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    return {"status": "success", **value}


@router.get("/{project_id}/plan/layers/lines", response_model=None)
async def plan_layer_lines(project_id: UUID, kind: str, request: Request) -> Response:
    """선 겹 GeoJSON(단층 · 지질구조선 · `summary` = 가닥 · 길이 m) — 범위 밖이면 받음."""
    if kind not in layers.LINE_KINDS:
        raise HTTPException(status_code=400, detail=f"없는 선 겹: {kind}")
    root, epsg, box = await _layer_root(project_id)
    processed = root / terrain.STAGE / "processed"
    try:
        got = await asyncio.to_thread(layers.ensure_lines, processed, box, epsg, kind)
    except RuntimeError as error:
        raise HTTPException(status_code=502, detail=str(error)) from error
    return cached_file_response(request, got["path"], "application/geo+json")


@router.get("/{project_id}/plan/layers/zero-order", response_model=None)
async def plan_layer_zero_order(
    project_id: UUID,
    min_m2: float = zero_order.ZERO_ORDER_MIN_M2,
    max_m2: float = zero_order.ZERO_ORDER_MAX_M2,
) -> dict[str, Any]:
    """0차곡 선(지형 추정 · 위 집수 `min_m2` ~ `max_m2`) — 격자 지문 · 임계 같으면 저장본."""
    if not 0 < min_m2 < max_m2:
        raise HTTPException(status_code=400, detail="0 < min_m2 < max_m2 여야 합니다.")
    root, epsg, _ = await _layer_root(project_id)
    try:
        value = await asyncio.to_thread(
            zero_order.zero_order_layer, root, terrain.models_dir_of(root), epsg, min_m2, max_m2
        )
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    return {"status": "success", **value}


@router.get("/{project_id}/plan/layers/forest-roads", response_model=None)
async def plan_layer_forest_roads(project_id: UUID) -> dict[str, Any]:
    """올린 임도망도(받은 범위로 자른 선 · 가닥 · 길이) — 올린 적 없으면 404."""
    root, epsg, box = await _layer_root(project_id)
    try:
        value = await asyncio.to_thread(forest_road.road_layer, root, box, epsg)
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    return {"status": "success", **value}


@router.post("/{project_id}/plan/layers/forest-roads", response_model=None)
async def plan_layer_forest_roads_upload(
    project_id: UUID, files: list[UploadFile] = File(...)
) -> dict[str, Any]:
    """임도망도 SHP 올리기(zip 하나 · shp · shx · dbf · prj 묶음 · 합계 100 MB) — 못 읽거나
    받은 범위 안 선이 없으면 400(옛 것 그대로) · 넘치면 413."""
    root, epsg, box = await _layer_root(project_id)
    got: list[tuple[str, bytes]] = []
    total = 0
    for upload in files:
        chunks = []
        while chunk := await upload.read(1024 * 1024):
            total += len(chunk)
            if total > forest_road.MAX_BYTES:
                raise HTTPException(
                    status_code=413,
                    detail=f"파일이 너무 큽니다 — 모두 {forest_road.MAX_BYTES // 2**20} MB 까지",
                )
            chunks.append(chunk)
        got.append((upload.filename or "", b"".join(chunks)))
    if not total:
        raise HTTPException(status_code=400, detail="빈 파일입니다.")
    try:
        value = await asyncio.to_thread(forest_road.save_roads, root, got, box, epsg)
    except ValueError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error
    return {"status": "success", **value}


@router.get("/{project_id}/plan/layers/landslide-zones", response_model=None)
async def plan_layer_landslide_zones(project_id: UUID) -> dict[str, Any]:
    """산사태취약지역(시도 표 · 점 + 지번 맞는 필지 테두리 · 표 없는 시도) — 지적도 못 받으면
    점만."""
    root, epsg, box = await _layer_root(project_id)
    processed = root / terrain.STAGE / "processed"
    try:
        got = await asyncio.to_thread(layers.ensure_cadastral, processed, box, epsg)
        cadastral = got["path"]
    except RuntimeError:
        cadastral = None
    value = await asyncio.to_thread(landslide_zone.zone_layer, root, cadastral, box, epsg)
    return {"status": "success", **value}