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` — 일의 진행률

범위 = 전체 시점 · 종점(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, HTTPException
from pydantic import BaseModel

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
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_Terrain as terrain

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["start"] or not plan["end"]:
        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["start"], plan["end"])]
    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},
    }