Aislo/D03_PlanAnalysis/D03_PlanAnalysis_Router.py

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

- `POST /api/projects/{id}/plan/terrain` — 지형 받기(뒤에서) · 끝나면 연도별 `surface_models` 행
- `GET …/plan/progress` — 일의 진행률

지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫). D03 화면에서 끈 지점(캐시 · PLAN 530-3)은
`points`(작업 좌표계 m)로 받아 그 지점으로 받음. 노선 정본(손 그은 선)은 `D03_PlanAnalysis_Network`
(PLAN 562-2 · 옛 추천 엔진 · 꺾임점 R 걷음).
"""

from __future__ import annotations

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

import aiomysql
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 PlanPoint(BaseModel):
    """끈 지점(작업 좌표계 m) — 연도 하나의 시점 · 종점."""

    year: int
    start: tuple[float, float]
    end: tuple[float, float]


class PointsBody(BaseModel):
    points: list[PlanPoint] | None = None


def _override(segments: list[dict[str, Any]], points: list[PlanPoint]) -> list[dict[str, Any]]:
    """연도 구간 지점을 끈 지점으로 — 연도 같음 · 직전 종점 = 다음 시점(1 mm) 아니면 거절."""
    by_year = {p.year: p for p in points}
    if set(by_year) != {seg["year"] for seg in segments}:
        raise HTTPException(status_code=400, detail="끈 지점의 연도가 사업연도와 다릅니다.")
    out = [
        {
            **seg,
            "start": tuple(by_year[seg["year"]].start),
            "end": tuple(by_year[seg["year"]].end),
            "saved": (seg["start"], seg["end"]),  # 저장 지점 — 그 분석본은 쌓임 거르기에서 남김
        }
        for seg in segments
    ]
    for prev, seg in zip(out, out[1:]):
        if max(abs(a - b) for a, b in zip(prev["end"], seg["start"])) > base.SAME_M:
            raise HTTPException(
                status_code=400, detail=f"{seg['year']}년 시점이 직전 종점과 다릅니다."
            )
    return out


async def _plan_inputs(
    project_id: UUID, points: list[PlanPoint] | None = None
) -> tuple[Path, dict[str, Any], list[dict[str, Any]]]:
    """(실경로, 프로젝트 칸, 연도 구간) — 빈 연도 · 좌표계 섞임 거절 · `points` = 끈 지점."""
    async with get_db_pool().acquire() as connection:
        stored = await get_project_storage_relative_path(connection, project_id)
        async with connection.cursor(aiomysql.DictCursor) as cursor:
            await cursor.execute(
                "SELECT road_type, design_speed_kph, route_start_m, route_end_m FROM projects"
                " WHERE id = %s",
                (str(project_id),),
            )
            project = await cursor.fetchone()
            await cursor.execute(
                "SELECT year, start_x, start_y, end_x, end_y, epsg, route_json, fingerprint "
                "FROM plan_route_years WHERE project_id = %s ORDER BY year",
                (str(project_id),),
            )
            rows = await cursor.fetchall()
    if not stored or not project:
        raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
    if not rows:
        raise HTTPException(
            status_code=400, detail="연도별 지점이 없습니다 — 1단계에서 지정하세요."
        )
    for row in rows:
        if any(row[k] is None for k in ("start_x", "start_y", "end_x", "end_y", "epsg")):
            raise HTTPException(status_code=400, detail=f"{row['year']}년 지점이 비었습니다.")
    if len({int(row["epsg"]) for row in rows}) != 1:
        raise HTTPException(status_code=400, detail="연도마다 좌표계가 다릅니다.")
    segments = [
        {
            "year": int(row["year"]),
            "start": (float(row["start_x"]), float(row["start_y"])),
            "end": (float(row["end_x"]), float(row["end_y"])),
            "epsg": int(row["epsg"]),
            "route_json": row["route_json"],
        }
        for row in rows
    ]
    if points:
        segments = _override(segments, points)
    return Path(resolve_stored_project_path(stored)), project, segments


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, years: dict[int, 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 models in years.values():
                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.post("/{project_id}/plan/terrain", response_model=None)
async def plan_terrain(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
    """지점 기준 수치지형도 · 위성 · 연도별 DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
    root, _, segments = await _plan_inputs(project_id, body.points if body else None)
    epsg = segments[0]["epsg"]

    async def work(report: Any) -> None:
        got = await asyncio.to_thread(terrain.ensure_plan_terrain, root, segments, epsg, report)
        await _save_models(project_id, got["years"], 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},
    }