Aislo/D03_PlanAnalysis/D03_PlanAnalysis_Router.py

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

- `POST /api/projects/{id}/plan/terrain` — 지형 받기(뒤에서) · 끝나면 연도별 `surface_models` 행
- `POST …/plan/analyze` — 연도 구간마다 노선 추천 갈래 넷(뒤에서 · 파생 저장본)
- `GET …/plan/progress` — 두 일의 진행률
- `GET …/plan/analysis` — 저장본 + 고른 갈래 · 이은 누가 · 이은 중심선
- `PUT …/plan/analysis` — [저장] 연도별 고른 갈래 → `plan_route_years.route_json` · `fingerprint`

지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫). D03 화면에서 끈 지점(캐시 · PLAN 530-3)은
`points`(작업 좌표계 m)로 받아 그 지점으로 받기 · 분석 · 보기 · [저장] 때만 D02 규칙으로 정본에 씀.
계산 자리 = ③ 서버 단독 — [저장]은 노선을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
"""

from __future__ import annotations

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

import aiomysql
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from common_util.common_util_auth import is_forest_service, require_company
from common_util.common_util_crs import tm_to_lonlat
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_Analyze as analyze
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


class ChoiceRequest(PointsBody):
    choices: dict[int, str]  # 연도 → 갈래


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)}
        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 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.post("/{project_id}/plan/analyze", response_model=None)
async def plan_analyze(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
    """연도 구간마다 노선 추천 갈래 넷(뒤에서 · 같은 입력이면 저장본)."""
    root, project, segments = await _plan_inputs(project_id, body.points if body else None)
    criteria = analyze.plan_criteria(project["road_type"], project["design_speed_kph"])
    missing = [s["year"] for s in segments if not terrain.models_dir_of(root, s["year"]).is_dir()]
    if missing:
        raise HTTPException(status_code=409, detail=f"지형을 먼저 받으세요: {missing}년")

    async def work(report: Any) -> None:
        await asyncio.to_thread(
            lambda: analyze.analyze_plan(root, segments, criteria, on_progress=report)
        )

    return _start_job(project_id, "analyze", 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},
    }


def _saved_mode(route_json: Any) -> str | None:
    try:
        value = json.loads(route_json) if isinstance(route_json, str) else route_json
    except ValueError:
        return None
    return value.get("mode") if isinstance(value, dict) else None


def _assemble(
    root: Path, project: dict[str, Any], segments: list[dict[str, Any]], choices: dict[int, Any]
) -> tuple[list[dict[str, Any] | None], list[dict[str, Any]] | None, list[list[float]]]:
    """(연도 저장본 · 이은 누가 · 이은 중심선) — 저장본이 하나라도 없으면 None · 빈 선."""
    results = [analyze.read_segment(root, seg) for seg in segments]
    if any(result is None for result in results):
        return results, None, []
    rows = analyze.chain(results, choices, float(project["route_start_m"] or 0.0))
    return results, rows, analyze.joined_line(results, rows)


def _query_points(points: str | None) -> list[PlanPoint] | None:
    if not points:
        return None
    try:
        return [PlanPoint(**item) for item in json.loads(points)]
    except (ValueError, TypeError) as error:
        raise HTTPException(status_code=400, detail="points 꼴이 다릅니다.") from error


@router.get("/{project_id}/plan/analysis", response_model=None)
async def plan_analysis(project_id: UUID, points: str | None = None) -> dict[str, Any]:
    """분석 저장본(지점이 지금과 같은 것만) + 고른 갈래(저장값 · 없으면 첫 성공 갈래).

    `points` = 끈 지점 JSON(`[{year, start, end}]` · 작업 좌표계 m) — 그 지점의 저장본을 봄.
    """
    root, project, segments = await _plan_inputs(project_id, _query_points(points))
    choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
    results, rows, line = await asyncio.to_thread(_assemble, root, project, segments, choices)
    return {
        "status": "success",
        "years": [
            {"year": seg["year"], "start": seg["start"], "end": seg["end"], "analysis": result}
            for seg, result in zip(segments, results)
        ],
        "chain": rows,
        "line": line,
        "start_chainage_m": float(project["route_start_m"] or 0.0),
    }


async def _store_points(project_id: UUID, segments: list[dict[str, Any]]) -> None:
    """끈 지점 → D02 정본 규칙(경위도 · 잇기 · 바뀐 연도만 · 그 연도 노선 비움)."""
    epsg = segments[0]["epsg"]

    def lonlat(xy: tuple[float, float]) -> base.LonLat:
        lon, lat = tm_to_lonlat(xy[0], xy[1], epsg)
        return base.LonLat(lon=lon, lat=lat)

    project, old = await base._load(project_id)
    start = int(project["project_year"])
    end = int(project["project_year_end"] or start)
    items = [
        base.YearPoints(year=s["year"], start=lonlat(s["start"]), end=lonlat(s["end"]))
        for s in segments
    ]
    new_epsg, rows = base.with_tm(base.chain_years(start, end, items))
    if new_epsg != epsg:
        raise HTTPException(
            status_code=409,
            detail="끈 지점이 다른 좌표계 띠로 넘어갔습니다 — 1단계에서 지정하세요.",
        )
    writes = [row for row in rows if base.changed(row, old.get(row["year"]), epsg)]
    if writes:
        await base._store(project_id, epsg, writes, [])


@router.put("/{project_id}/plan/analysis", response_model=None)
async def save_plan_analysis(
    project_id: UUID,
    request: ChoiceRequest,
    session: dict[str, Any] = Depends(require_company),
) -> dict[str, Any]:
    """[저장] — 끈 지점(있으면 정본으로) · 연도별 고른 갈래를 서버 저장본으로 다시 이어 씀."""
    if request.points and not is_forest_service(session):
        raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.")
    root, project, segments = await _plan_inputs(project_id, request.points)
    bad = [m for m in request.choices.values() if m not in analyze.PLAN_MODES]
    if bad:
        raise HTTPException(status_code=400, detail=f"모르는 갈래: {bad}")
    choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
    choices.update(request.choices)
    results, rows, _ = await asyncio.to_thread(_assemble, root, project, segments, choices)
    if rows is None:
        raise HTTPException(
            status_code=409, detail="분석 저장본이 없습니다 — [분석]을 먼저 하세요."
        )
    lost = [r["year"] for r in rows if choices.get(r["year"]) not in (None, r["mode"])]
    if lost:
        raise HTTPException(status_code=409, detail=f"고른 갈래에 노선이 없습니다: {lost}년")
    if request.points:
        await _store_points(project_id, segments)
    async with get_db_pool().acquire() as connection:
        await connection.begin()
        try:
            async with connection.cursor() as cursor:
                for result, row in zip(results, rows):
                    route = result["routes"][row["mode"]]
                    body = {**row, **{k: route[k] for k in ("planned", "nodes", "curves")}}
                    await cursor.execute(
                        "UPDATE plan_route_years SET route_json = %s, fingerprint = %s "
                        "WHERE project_id = %s AND year = %s",
                        (
                            json.dumps(body, ensure_ascii=False),
                            result["fingerprint"],
                            str(project_id),
                            row["year"],
                        ),
                    )
            await connection.commit()
        except Exception:
            await connection.rollback()
            raise
    return {"status": "success", "chain": rows}