"""D 예정노선 2단계 분석 — 연도 구간마다 B05 노선 추천 엔진으로 지점을 이음 (요구 345 · PLAN 530-2).

- 구간 = [P(i-1), P(i)] 한 번씩 `generate_initial_route` — 시종점을 옮기지 않아 연도 경계가 맞물림
- 갈래 넷(`PLAN_MODES` · follow 뺌 — 두 점이면 직선) · 연도마다 고름
- 추천 기준 = 프로젝트 임도 종류 · 설계속도 → B 법정 함수 · 지형 구분은 B 폴백(`TERRAIN_TYPE`)
- 계류 이격 = 도엽 하천선(B 로더 그대로) · 복도 반폭 = `PLAN_CORRIDOR_M`(D 만)
- 연도 경계는 꺾인 채(경계점 = 측점) · 누가거리 = 등록 시작 누가부터 고른 갈래 길이로 이어 셈
- 파생 저장본 = `D03_PlanAnalysis/_derived/route_<연도>_<지점 열쇠>.json`(입력 지문 같으면 다시
  안 셈) · 끈 지점 분석이 저장 지점 분석본을 덮지 않음 · 연도마다 최근 `KEEP_PER_YEAR` 개 + 저장
  지점 것만 남김 · 옛 꼴 `route_<연도>.json` 은 읽기만(공유 storage — 옛 판 서버가 아직 읽음)

계산 자리 = ③ 서버 단독(B 추천과 같음).

⚠ 파이썬·TS 짝 — `pick_mode` · `chain` · `joined_line` · `with_edits` ↔
`logic/D03_PlanAnalysis_Plan.ts` `pickMode` · `chainRows` · `joinedLine` · `withEdits`(화면이 고르는
즉시) · 거울 시험 `test_530_3_plan_logic.py` · `test_543_plan_edit.py`.
"""

from __future__ import annotations

import hashlib
import json
import logging
import math
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Callable

import numpy as np
import shapely
from shapely.geometry import LineString, MultiLineString

from B05_Profile.B05_Profile_Engine_Grade import (
    legal_grade_limit_pct,
    legal_plan_radius_min_m,
    resolve_design_speed,
)
from B05_Profile.B05_Profile_Engine_RouteInitial import (
    _POOLS,
    _TERRAINS,
    generate_initial_route,
    load_terrain_and_streams,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint
from common_util.common_util_json import atomic_write_json
from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER
from config.config_system_terrain import PLAN_ANALYZE_WORKERS
from D03_PlanAnalysis.D03_PlanAnalysis_Terrain import PLAN_CORRIDOR_M, models_dir_of

logger = logging.getLogger(__name__)

PLAN_MODES = ("gradeline", "earthwork", "gentle", "weighted")
# 도엽 DEM(스무딩) — B 「수치지형도」 분석 지형과 같은 고르기
SELECTION = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
# D 등록에 지형 구분 칸 없음 — B `_road_settings` 못 읽을 때 폴백 그대로(가장 완화)
TERRAIN_TYPE = "special"
ALGO = 1
# 계류 거르기 칸 · 지형 상자 여유(m) — `near_streams`
STREAM_TILE_M = 200.0
STREAM_PAD_M = 100.0
# 연도마다 남길 분석본 수 — 저장 지점 분석본은 따로 늘 남김
KEEP_PER_YEAR = 3

Progress = Callable[[int, str], None]
Loader = Callable[..., tuple[Any, list]]


def derived_dir(project_root: Path) -> Path:
    return project_root / "D03_PlanAnalysis" / "_derived"


def plan_criteria(road_type: str | None, design_speed_kph: int | None) -> dict[str, Any]:
    """추천 기준 — 종단 상한(비율) · 최소 R · 설계속도(B 법정 함수)."""
    grade_class = str(road_type or "work")
    speed = resolve_design_speed(grade_class, design_speed_kph)
    return {
        "max_grade": legal_grade_limit_pct(grade_class, TERRAIN_TYPE, False, speed) / 100.0,
        "min_radius_m": legal_plan_radius_min_m(speed, TERRAIN_TYPE),
        "design_speed": speed,
    }


def _point(value: Any) -> tuple[float, float]:
    """엔진 출력 자리수(0.1 mm)로 — 노선 끝점 = 지점 그대로 · 연도 경계가 정확히 맞물림."""
    return (round(float(value[0]), 4), round(float(value[1]), 4))


def _same_point(a: Any, b: Any) -> bool:
    return abs(float(a[0]) - float(b[0])) < 1e-3 and abs(float(a[1]) - float(b[1])) < 1e-3


def segment_path(project_root: Path, seg: dict[str, Any]) -> Path:
    """연도 · 지점 열쇠 파일 — 지점이 다르면 다른 파일(덮지 않음)."""
    key = hashlib.blake2b(
        repr((_point(seg["start"]), _point(seg["end"]))).encode(), digest_size=4
    ).hexdigest()
    return derived_dir(project_root) / f"route_{int(seg['year'])}_{key}.json"


def _find(project_root: Path, seg: dict[str, Any]) -> tuple[Path, dict[str, Any]] | None:
    """지점이 지금과 같은 저장본(1 mm) — 열쇠 파일 · 같은 연도 다른 열쇠(새것부터) · 옛 꼴 차례."""
    year = int(seg["year"])
    folder = derived_dir(project_root)
    exact = segment_path(project_root, seg)
    others = sorted(
        (p for p in folder.glob(f"route_{year}_*.json") if p != exact),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    for path in (exact, *others, folder / f"route_{year}.json"):
        try:
            stored = json.loads(path.read_text("utf-8"))
        except (OSError, ValueError):
            continue
        if _same_point(stored["start"], seg["start"]) and _same_point(stored["end"], seg["end"]):
            return path, stored
    return None


def read_segment(project_root: Path, seg: dict[str, Any]) -> dict[str, Any] | None:
    """저장본 — 지점이 지금과 같을 때만(입력 지문은 분석 때 다시 봄)."""
    found = _find(project_root, seg)
    return found[1] if found else None


def _prune(project_root: Path, seg: dict[str, Any]) -> None:
    """연도 분석본 — 최근 `KEEP_PER_YEAR` 개 + 저장 지점(`saved` · 없으면 seg) 것만 남김 · 옛 꼴은
    안 지움."""
    start, end = seg.get("saved") or (seg["start"], seg["end"])
    canon = _find(project_root, {"year": seg["year"], "start": start, "end": end})
    files = sorted(
        derived_dir(project_root).glob(f"route_{int(seg['year'])}_*.json"),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    for path in files[KEEP_PER_YEAR:]:
        if canon is None or path != canon[0]:
            path.unlink(missing_ok=True)


def _route_view(result: dict[str, Any]) -> dict[str, Any]:
    keep = ("planned", "nodes", "curves", "metrics", "relaxed", "terrain_bound_pct")
    return {key: result.get(key) for key in keep}


def _engine_inputs(criteria: dict[str, Any]) -> dict[str, Any]:
    return {
        "weights": None,
        "max_grade": criteria["max_grade"],
        "min_radius_m": criteria["min_radius_m"],
        "stream_offset_m": None,
        "grade_smooth_m": None,
        "target_grade": None,
        "with_profile": False,
        "design_speed": criteria["design_speed"],
    }


def segment_fingerprint(
    seg: dict[str, Any],
    terrain: Any,
    streams: list,
    criteria: dict[str, Any],
    corridor_m: float = PLAN_CORRIDOR_M,
) -> str:
    """구간 입력 지문 — 갈래 넷 B 저장 키 · 복도 · 판."""
    expected = [_point(seg["start"]), _point(seg["end"])]
    inputs = _engine_inputs(criteria)
    keys = [
        request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0]
        for mode in PLAN_MODES
    ]
    return hashlib.blake2b(
        repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16
    ).hexdigest()


def analyze_segment(
    project_root: Path,
    seg: dict[str, Any],
    terrain: Any,
    streams: list,
    criteria: dict[str, Any],
    corridor_m: float = PLAN_CORRIDOR_M,
) -> dict[str, Any]:
    """연도 구간 하나 · 갈래 넷 — 입력 지문이 저장본과 같으면 그대로(`stored` True)."""
    expected = [_point(seg["start"]), _point(seg["end"])]
    inputs = _engine_inputs(criteria)
    fingerprint = segment_fingerprint(seg, terrain, streams, criteria, corridor_m)
    stored = read_segment(project_root, seg)
    if stored is not None and stored.get("fingerprint") == fingerprint:
        return {**stored, "stored": True}
    routes: dict[str, Any] = {}
    for mode in PLAN_MODES:
        try:
            routes[mode] = _route_view(
                generate_initial_route(
                    expected, terrain, streams, mode=mode, corridor_m=corridor_m, **inputs
                )
            )
        except ValueError as error:  # NoRoute 포함 — 그 갈래만 빔
            routes[mode] = {"error": str(error)}
    result = {
        "year": int(seg["year"]),
        "start": list(expected[0]),
        "end": list(expected[1]),
        "fingerprint": fingerprint,
        "corridor_m": float(corridor_m),
        "criteria": {
            "max_grade_pct": round(criteria["max_grade"] * 100.0, 2),
            "min_radius_m": criteria["min_radius_m"],
            "design_speed_kph": criteria["design_speed"],
            "terrain_type": TERRAIN_TYPE,
        },
        "routes": routes,
    }
    atomic_write_json(segment_path(project_root, seg), result)
    _prune(project_root, seg)
    return {**result, "stored": False}


def near_streams(terrain: Any, streams: list) -> list:
    """지형 상자 안 어느 점에서든 가장 가까운 계류일 수 있는 선만 — 나머지는 엔진 값에 안 닿음.

    칸마다 닿는 거리 = 칸 가운데 계류 거리 + 칸 반대각(거리는 1-립시츠) — 그 안에 드는 선만 남기면
    상자 안 점의 계류 거리 · 건너기 · 격자 거리가 통째와 같음(통째 거리 = 추천 시간의 큰 몫).
    """
    if not streams:
        return streams
    x0, x1 = float(np.min(terrain.x)) - STREAM_PAD_M, float(np.max(terrain.x)) + STREAM_PAD_M
    y0, y1 = float(np.min(terrain.y)) - STREAM_PAD_M, float(np.max(terrain.y)) + STREAM_PAD_M
    xs = np.linspace(x0, x1, max(1, math.ceil((x1 - x0) / STREAM_TILE_M)) + 1)
    ys = np.linspace(y0, y1, max(1, math.ceil((y1 - y0) / STREAM_TILE_M)) + 1)
    lo_x, lo_y = np.meshgrid(xs[:-1], ys[:-1])
    hi_x, hi_y = np.meshgrid(xs[1:], ys[1:])
    tiles = shapely.box(lo_x.ravel(), lo_y.ravel(), hi_x.ravel(), hi_y.ravel())
    reach = shapely.distance(shapely.centroid(tiles), MultiLineString(streams)) + 0.5 * math.hypot(
        xs[1] - xs[0], ys[1] - ys[0]
    )
    lines = np.array([LineString(line) for line in streams], dtype=object)
    keep = (shapely.distance(lines[:, None], tiles[None, :]) <= reach[None, :]).any(axis=1)
    return [line for line, kept in zip(streams, keep) if kept]


def _load(project_root: Path, year: int, loader: Loader) -> tuple[Any, list]:
    terrain, streams = loader(
        project_root,
        SELECTION,
        models_dir_of(project_root, year),
        derived_dir(project_root) / f"cost_{year}",
    )
    return terrain, near_streams(terrain, streams)


def _segment_job(
    project_root: Path,
    seg: dict[str, Any],
    criteria: dict[str, Any],
    corridor_m: float,
    loader: Loader,
) -> dict[str, Any]:
    """연도 구간 하나 — 지형 읽기부터(제 프로세스면 끝날 때 비용면 · 묶음 메모리를 놓음)."""
    terrain, streams = _load(project_root, int(seg["year"]), loader)
    return analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m)


def _worker_job(*args: Any) -> dict[str, Any]:
    """나란히 셈 프로세스 — 구간 하나 뒤 B 기억(비용면 · 묶음)을 비워 다음 구간 메모리를 놓음."""
    try:
        return _segment_job(*args)
    finally:
        _POOLS.clear()
        _TERRAINS.clear()


def analyze_plan(
    project_root: Path,
    segments: list[dict[str, Any]],
    criteria: dict[str, Any],
    *,
    loader: Loader = load_terrain_and_streams,
    corridor_m: float = PLAN_CORRIDOR_M,
    on_progress: Progress | None = None,
    workers: int = PLAN_ANALYZE_WORKERS,
) -> list[dict[str, Any]]:
    """연도 구간마다(연도 DEM · 연도 비용면 캐시) `analyze_segment`.

    다시 셀 구간이 둘 이상이면 프로세스 `workers` 개로 나란히 — 구간끼리 입력 · 저장본이 따로라
    결과 같음. `workers` 1 = 이 프로세스(가짜 지형 시험)."""

    def report(percent: int, message: str) -> None:
        if on_progress is not None:
            on_progress(percent, message)

    results: list[dict[str, Any] | None] = [None] * len(segments)
    todo = []
    for i, seg in enumerate(segments):
        stored = read_segment(project_root, seg)
        if stored is not None:  # 저장본이 있을 때만 이 프로세스에서 지형을 읽어 지문을 봄
            terrain, streams = _load(project_root, int(seg["year"]), loader)
            if stored.get("fingerprint") == segment_fingerprint(
                seg, terrain, streams, criteria, corridor_m
            ):
                results[i] = {**stored, "stored": True}
                continue
        todo.append(i)

    count = min(workers, len(todo))
    if count <= 1:
        for n, i in enumerate(todo):
            report(100 * n // len(segments), f"{int(segments[i]['year'])}년 구간 노선 추천 중")
            results[i] = _segment_job(project_root, segments[i], criteria, corridor_m, loader)
    else:
        years = "·".join(str(int(segments[i]["year"])) for i in todo)
        report(0, f"{years}년 구간 노선 추천 중(나란히 {count})")
        context = multiprocessing.get_context("spawn")
        with ProcessPoolExecutor(count, mp_context=context) as pool:
            futures = {
                pool.submit(_worker_job, project_root, segments[i], criteria, corridor_m, loader): i
                for i in todo
            }
            for n, future in enumerate(as_completed(futures), 1):
                i = futures[future]
                results[i] = future.result()
                report(
                    100 * n // len(todo), f"{int(segments[i]['year'])}년 구간 끝 ({n}/{len(todo)})"
                )
    report(100, "분석 끝")
    return results  # type: ignore[return-value]


def pick_mode(result: dict[str, Any], wanted: str | None) -> str | None:
    """고른 갈래 — 없거나 빈 갈래면 차례대로 첫 성공 갈래."""
    routes = result.get("routes") or {}
    if wanted in routes and "error" not in routes[wanted]:
        return wanted
    return next((m for m in PLAN_MODES if m in routes and "error" not in routes[m]), None)


def chain(
    results: list[dict[str, Any]], choices: dict[int, str | None], start_chainage_m: float
) -> list[dict[str, Any]]:
    """연도별 고른 갈래 · 시작 · 끝 누가(이어 셈) — 앞 연도 끝 누가 = 다음 연도 시작 누가."""
    at = float(start_chainage_m or 0.0)
    rows = []
    for result in results:
        mode = pick_mode(result, choices.get(int(result["year"])))
        length = float(result["routes"][mode]["metrics"]["length_m"]) if mode else 0.0
        rows.append(
            {
                "year": int(result["year"]),
                "mode": mode,
                "length_m": round(length, 3),
                "start_chainage_m": round(at, 3),
                "end_chainage_m": round(at + length, 3),
            }
        )
        at += length
    return rows


def line_length(line: list) -> float:
    """선 길이(m) — 짝: `lineLength`(같은 차례로 더함)."""
    total = 0.0
    for a, b in zip(line, line[1:]):
        total += math.hypot(float(b[0]) - float(a[0]), float(b[1]) - float(a[1]))
    return total


def with_edits(
    results: list[dict[str, Any]], edits: dict[int, dict[str, Any]]
) -> list[dict[str, Any]]:
    """고친 노선(`{mode, planned, metrics?}`)을 그 갈래 자리에 갈아 끼운 분석 사본 — 길이 = 고친 선
    길이. 그 갈래가 없거나 빈 갈래면 그대로."""
    out = []
    for result in results:
        edit = edits.get(int(result["year"]))
        route = (result.get("routes") or {}).get(edit["mode"]) if edit else None
        if not edit or route is None or "error" in route:
            out.append(result)
            continue
        metrics = {**(edit.get("metrics") or {}), "length_m": line_length(edit["planned"])}
        routes = {
            **result["routes"],
            edit["mode"]: {**route, "planned": edit["planned"], "metrics": metrics, "edited": True},
        }
        out.append({**result, "routes": routes})
    return out


def joined_line(results: list[dict[str, Any]], rows: list[dict[str, Any]]) -> list[list[float]]:
    """고른 갈래를 이은 중심선 — 경계점 한 번만(꺾인 채)."""
    line: list[list[float]] = []
    for result, row in zip(results, rows):
        if row["mode"] is None:
            continue
        planned = result["routes"][row["mode"]]["planned"]
        line.extend(planned[1:] if line and _same_point(line[-1], planned[0]) else planned)
    return line