"""E03 2D 측점 해석 API (요구 345 · 347 · 351 · PLAN 532-3).

- `POST /api/projects/{id}/stab/analyze2d` — 전 측점 해석(뒤에서 · 바뀐 측점만) · 같은 지문 = 안 셈
- `GET …/stab/progress` — 진행률
- `GET …/stab/result2d` — 요약 + 측점 목록(판정 · 대표 단면 · 벽 수 · 지배 비탈) + 횡단 없는 측점
- `GET …/stab/section/{station_key}` — 그 측점 지층 폴리곤 · 지하수선 · 임계면 · 후보 · 절편 · 격자
- `PUT …/stab/representative` — 대표 단면 켬/끔 → 작업본 `E03_StabSection/representative.json`
- `GET …/stab/corridor3d` — 복사본 노선면(1 m 솎음 · 3D 색칠 보기용)

입력 = E01 복사본(측점 · 횡단 · DTM) + E02 지반 조건(작업본 · 없으면 초기값)
계산 자리 = ③ 서버 단독.
결과 = 파생 저장본 `_derived/stab2d_station/<측점 지문>` · `_derived/stab2d_summary/<전체 지문>` —
측점 지문 = 엔진 코드 · 복사 지문 · 물성 · 조건 · 하중 · 지하수 갈래 · 그 측점 지층 · 벽.
공유 회사(보기만)는 `require_project_access` 가 POST · PUT 를 막음.
"""

from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import logging
import math
import os
import subprocess
import sys
import tempfile
import time
from functools import lru_cache
from pathlib import Path
from typing import Any
from uuid import UUID

import numpy as np
import shapely
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from scipy.spatial import cKDTree

from common_util.common_util_auth import verify_session
from common_util.common_util_derived_store import (
    derived_dir,
    derived_key,
    read_derived,
    write_derived,
)
from common_util.common_util_json import atomic_write_json
from common_util.common_util_station_key import chainage_key
from common_util.common_util_surface_sampler import DtmGridSampler
from E01_StabRegister import E01_StabRegister_Source as src
from E02_StabGround import E02_StabGround_Model as ground_model
from E02_StabGround import E02_StabGround_Router as e02
from E03_StabSection import E03_StabSection_View as view
from E03_StabSection import E03_StabSection_Walls as walls_mod
from E03_StabSection import E03_StabSection_Worker as worker
from E03_StabSection.E03_StabSection_Engine_Model import (
    ground_from_cross,
    section_model_from_b06,
    widen_ground,
)

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["E03 Stab Section"])

PROJECT_ROOT = Path(__file__).resolve().parents[1]
KIND_STATION = worker.KIND_STATION
KIND_SUMMARY = "stab2d_summary"
REPRESENTATIVE = Path("E03_StabSection") / "representative.json"
ENGINE_FILES = (
    "E03_StabSection_Types.py",
    "E03_StabSection_Engine_Model.py",
    "E03_StabSection_Engine_LEM.py",
    "E03_StabSection_Engine_Search.py",
    "E03_StabSection_Engine_Station.py",
)
#: 3D 보기 노선면 솎음 간격(m)
CORRIDOR_STEP_M = 1.0
TERRAIN_MARGIN_M = 80.0  # 노선면 둘레 지형 띠 폭(B05 지형 띠 여유와 같음)
TERRAIN_STEP_M = 2.0  # 지형 칸(이보다 촘촘히 안 함)
TERRAIN_MAX_CELLS = 250_000  # 응답 크기 상한 — 넘으면 칸을 키움
#: 프로세스 수 — None = 코어 − 2 · 1 = 서버 안에서 차례로(시험)
WORKERS: int | None = None

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


class RepresentativeRequest(BaseModel):
    overrides: dict[str, bool | None]


@lru_cache(maxsize=1)
def engine_hash() -> str:
    """엔진 코드 지문 — 코드가 바뀌면 모든 측점을 다시 셈."""
    digest = hashlib.sha1()
    for name in ENGINE_FILES:
        digest.update((Path(__file__).parent / name).read_bytes())
    return digest.hexdigest()[:12]


def _ground(root: Path) -> dict[str, Any]:
    """E02 지반 조건 — 작업본 · 없으면 초기값(옛 정본 = 빠진 지하수 칸 기본값 · E02 화면과 같음)."""
    ground = e02._read_working(root) or e02.ensure_initial(root)
    return ground_model.upgrade(ground, ground_model.load_defaults())


def _layers(resolved: dict[str, Any] | None) -> list | None:
    """E02 측점 적용 지층 → 엔진 지층(위부터 [물성 id, 원지반 아래 바닥 깊이 | None]).

    없으면 None(엔진 기본 = B06 라벨)."""
    if not resolved or not resolved.get("upper"):
        return None
    if resolved.get("lower"):
        return [[resolved["upper"], float(resolved["soil_thickness_m"])], [resolved["lower"], None]]
    return [[resolved["upper"], None]]


@lru_cache(maxsize=4)
def _walls(source: str, _copy_fp: str) -> dict[str, dict[str, Any]]:
    """측점별 벽(532-4 · Node 한 번 약 2 s) — 복사본은 동결이라 복사 지문으로 기억."""
    return walls_mod.walls_by_station(source)


def plan_run(root: Path, ground: dict[str, Any]) -> dict[str, Any]:
    """측점마다 입력 · 지문 + 전체 지문(판정 기준 포함) — 복사본이 없으면 400."""
    try:
        source = src.source_dir(root)
        copy_fp = src.manifest(source)["source_fingerprint"]
        stations = src.stations(source)
    except (OSError, KeyError) as error:
        raise HTTPException(status_code=400, detail="원본 복사본이 없습니다.") from error
    seismic = ground.get("seismic") or {}
    conditions = ["dry", "wet"] + (["seismic"] if seismic.get("enabled") else [])
    shared = {
        "materials": ground.get("materials"),
        "conditions": conditions,
        "seismic_kh": float(seismic.get("kh") or 0.0) if seismic.get("enabled") else 0.0,
        "road_surcharge_kpa": float(ground.get("road_surcharge_kpa") or 0.0),
        "groundwater": ground.get("groundwater"),
    }
    resolved = {chainage_key(r["chainage_m"]): r for r in ground_model.resolve_stations(ground)}
    try:
        found = _walls(str(source), copy_fp)
    except RuntimeError as error:
        raise HTTPException(status_code=500, detail=str(error)) from error
    items = []
    for station in stations:
        key = chainage_key(station["chainage_m"])
        layers = _layers(resolved.get(key))
        entry = found.get(key)
        parts = [engine_hash(), copy_fp, shared, key, layers] + ([entry] if entry else [])
        items.append(
            {
                "key": key,
                "chainage_m": station["chainage_m"],
                "label": station.get("label"),
                "ground_type": station.get("ground_type"),
                "center": [station.get("center_x"), station.get("center_y")],
                "layers": layers,
                "walls": entry,
                "fp": derived_key(*parts),  # 벽 없는 측점 = 종전 지문 꼴
            }
        )
    factors = ground.get("safety_factors") or {}
    overall = derived_key([item["fp"] for item in items], factors)
    return {
        "source": source,
        "ground": ground,
        "shared": shared,
        "items": items,
        "factors": factors,
        "fp": overall,
    }


def _station_input(source: Path, item: dict[str, Any], sampler: Any) -> dict[str, Any]:
    """측점 하나 엔진 입력 — 횡단 · 설계 · 넓힌 지반선(DTM 밖은 표본 그대로)."""
    cross = src.cross_section(source, item["chainage_m"])
    try:
        ground = widen_ground(cross, sampler) if sampler else ground_from_cross(cross)
    except (KeyError, TypeError, ValueError):
        ground = ground_from_cross(cross)
    return {
        "cross": cross,
        "design": walls_mod.design_with_walls(cross["design"], item.get("walls")),
        "walls": walls_mod.walls_of(item.get("walls")),
        "ground": [list(p) for p in ground],
    }


def _sampler(source: Path) -> Any:
    try:
        return DtmGridSampler.from_npz(src.terrain_grid_path(source))
    except (OSError, FileNotFoundError, ValueError):
        logger.warning("E03 지형 격자 없음 — 횡단 표본 폭으로 해석: %s", source)
        return None


def _meta(item: dict[str, Any], design: dict[str, Any]) -> dict[str, Any]:
    keys = ("key", "chainage_m", "label", "ground_type", "center", "fp")
    walls = len(walls_mod.walls_of(item.get("walls")) or [])
    return {
        **{k: item[k] for k in keys},
        "section_mode": design.get("section_mode"),
        "walls": walls,
    }


def _run_worker(job: dict[str, Any], emit: Any) -> None:
    """일꾼 프로세스(`E03_StabSection_Worker`) — 한 줄마다 emit · 실패면 RuntimeError.

    stderr 는 파일로(파이프로 두면 경고가 쌓여 일꾼이 멈춤)."""
    with tempfile.TemporaryDirectory(prefix="aislo_e03_") as folder:
        path = Path(folder) / "job.json"
        path.write_text(json.dumps(job, ensure_ascii=False), encoding="utf-8")
        log = Path(folder) / "stderr.txt"
        with log.open("w", encoding="utf-8") as errors:
            process = subprocess.Popen(
                [sys.executable, "-m", "E03_StabSection.E03_StabSection_Worker", str(path)],
                cwd=PROJECT_ROOT,
                stdout=subprocess.PIPE,
                stderr=errors,
                text=True,
                encoding="utf-8",
            )
            for line in process.stdout:
                if line.startswith("{"):
                    emit(json.loads(line).get("done"))
            code = process.wait()
        if code != 0:
            tail = log.read_text(encoding="utf-8", errors="replace").strip().splitlines()[-1:]
            raise RuntimeError(f"해석 일꾼 실패: {tail}")


def run_all(root: Path, plan: dict[str, Any], report: Any) -> int:
    """지문이 없는 측점만 셈(일꾼 프로세스 병렬) → 측점 저장본 · 요약 저장본. 센 측점 수."""
    source, shared, items = plan["source"], plan["shared"], plan["items"]
    todo = [it for it in items if read_derived(root, KIND_STATION, it["fp"]) is None]
    if todo:
        sampler = _sampler(source)
        job_items = []
        for it in todo:
            data = _station_input(source, it, sampler)
            job_items.append({"fp": it["fp"], "meta": _meta(it, data["design"]), **data})
            job_items[-1]["layers"] = it["layers"]
        job = {
            "root": str(root),
            "keep": max(3 * len(items), 6),
            "workers": WORKERS or max(1, (os.cpu_count() or 2) - 2),
            "conditions_ground": plan["ground"],
            "kwargs": {
                "materials": shared["materials"],
                "conditions": list(shared["conditions"]),
                "seismic_kh": shared["seismic_kh"],
                "road_surcharge_kpa": shared["road_surcharge_kpa"],
            },
            "items": job_items,
        }
        report(2, f"측점 {len(todo)} 개 해석 시작")
        done = [0]

        def emit(_key: str | None) -> None:
            done[0] += 1
            report(2 + 95 * done[0] // len(todo), f"측점 {done[0]} / {len(todo)}")

        if WORKERS == 1:
            worker.run_job(job, emit)
        else:
            _run_worker(job, emit)
    rows = []
    for it in items:
        saved = read_derived(root, KIND_STATION, it["fp"])
        value = saved["value"] if saved else {"meta": _meta(it, {}), "out": None, "error": "없음"}
        rows.append(view.station_row(value["meta"], value["out"], plan["factors"], value["error"]))
    summary = {
        "fingerprint": plan["fp"],
        "computed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "conditions": shared["conditions"],
        "method": view.PRIMARY,
        "factors": plan["factors"],
        "summary": view.summarize(rows, shared["conditions"]),
        "stations": rows,
    }
    write_derived(root, KIND_SUMMARY, plan["fp"], summary)
    return len(todo)


def _latest_summary(root: Path, fp: str) -> tuple[dict[str, Any] | None, bool]:
    """(요약, 지금 입력과 같은지) — 같은 지문이 없으면 가장 최근 것(옛 결과)."""
    saved = read_derived(root, KIND_SUMMARY, fp)
    if saved:
        return saved["value"], True
    files = sorted(
        derived_dir(root, KIND_SUMMARY).glob("*.json"),
        key=lambda p: p.stat().st_mtime_ns,
        reverse=True,
    )
    for path in files:
        old = read_derived(root, KIND_SUMMARY, path.stem)
        if old:
            return old["value"], False
    return None, False


def _overrides(root: Path) -> dict[str, bool]:
    path = root / REPRESENTATIVE
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return {k: bool(v) for k, v in (data.get("overrides") or {}).items() if v is not None}


@router.post("/{project_id}/stab/analyze2d", response_model=None)
async def analyze2d(project_id: UUID) -> dict[str, Any]:
    """전 측점 해석(뒤에서) — 지문이 같으면 다시 안 셈(`fresh`) · 진행은 `GET …/stab/progress`."""
    root, _ = await e02._project(project_id)
    plan = plan_run(root, _ground(root))
    if read_derived(root, KIND_SUMMARY, plan["fp"]):
        return {"status": "fresh", "fingerprint": plan["fp"]}
    key = str(project_id)
    if (_JOBS.get(key) or {}).get("running"):
        raise HTTPException(status_code=409, detail="이미 해석 중입니다.")
    job = {"running": True, "percent": 0, "message": "시작", "error": None, "computed": 0}
    _JOBS[key] = job

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

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

    task = asyncio.create_task(work(), name=f"E03 analyze2d {project_id}")
    _TASKS.add(task)
    task.add_done_callback(_TASKS.discard)
    return {"status": "accepted", "fingerprint": plan["fp"], **job}


@router.get("/{project_id}/stab/progress", response_model=None)
async def progress(project_id: UUID) -> dict[str, Any]:
    await e02._project(project_id)
    job = _JOBS.get(str(project_id))
    return dict(job) if job else {"running": False, "percent": 0, "message": "", "error": None}


@router.get("/{project_id}/stab/result2d", response_model=None)
async def result2d(
    project_id: UUID, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    """요약 + 측점 목록 — 결과가 없으면 복사본 측점만(status none)."""
    root, project = await e02._project(project_id)
    ground = _ground(root)
    plan = plan_run(root, ground)
    summary, fresh = _latest_summary(root, plan["fp"])
    editable = e02._editable(session, project)
    running = bool((_JOBS.get(str(project_id)) or {}).get("running"))
    missing = src.missing_stations(plan["source"])  # 횡단 없어 해석에서 빠진 측점(화면 띠)
    if not summary:
        rows = [view.station_row(_meta(item, {}), None, plan["factors"]) for item in plan["items"]]
        view.mark_representative(rows, _overrides(root))
        rows = [{**r, **view.governing_row(r)} for r in rows]
        return {
            "status": "none",
            "fresh": False,
            "editable": editable,
            "running": running,
            "conditions": plan["shared"]["conditions"],
            "factors": plan["factors"],
            "stations": rows,
            "missing_stations": missing,
        }
    rows = summary["stations"]
    view.mark_representative(rows, _overrides(root))
    rows = [{**r, **view.governing_row(r)} for r in rows]
    return {
        "status": "ok",
        "fresh": fresh,
        "editable": editable,
        "running": running,
        **{k: v for k, v in summary.items() if k != "stations"},
        "stations": rows,
        "missing_stations": missing,
    }


@router.get("/{project_id}/stab/section/{station_key}", response_model=None)
async def section(project_id: UUID, station_key: str) -> dict[str, Any]:
    """측점 하나 — 해석에 쓴 지반선 · 지층으로 모형을 다시 세워(빠름) 결과와 함께."""
    root, _ = await e02._project(project_id)
    ground = _ground(root)
    plan = plan_run(root, ground)
    summary, fresh = _latest_summary(root, plan["fp"])
    row = next((r for r in (summary or {}).get("stations", []) if r["key"] == station_key), None)
    saved = read_derived(root, KIND_STATION, row["fp"]) if row else None
    if not saved or not saved["value"]["out"]:
        raise HTTPException(status_code=404, detail="그 측점 해석 결과가 없습니다.")
    value = saved["value"]
    cross = src.cross_section(plan["source"], row["chainage_m"])
    entry = next((it["walls"] for it in plan["items"] if it["key"] == station_key), None)
    design = walls_mod.design_with_walls(cross["design"], entry)
    kw = {
        "materials": plan["shared"]["materials"],
        "layers": value["inputs"]["layers"],
        "ground": [tuple(p) for p in value["inputs"]["ground"]],
        "road_surcharge_kpa": plan["shared"]["road_surcharge_kpa"],
        "walls": walls_mod.walls_of(entry),
    }
    dry = section_model_from_b06(cross, design, **kw)
    water, rules = {}, {}
    for cond in summary["conditions"]:
        # 수위선 = E02 고른 갈래(r_u · 없음 = 선 없음)
        model = section_model_from_b06(cross, design, wet=cond == "wet", **kw)
        gw = ground_model.station_groundwater(ground, cond, model)
        water[cond] = gw.get("line") if gw.get("mode") == "line" else None
        rules[cond] = {"mode": gw.get("mode"), "ru": gw.get("ru")}
    out = view.section_view(row, value["out"], dry, water)
    out["water_rule"] = rules
    out["fresh"] = fresh
    return out


@router.put("/{project_id}/stab/representative", response_model=None)
async def put_representative(project_id: UUID, payload: RepresentativeRequest) -> dict[str, Any]:
    """대표 단면 켬/끔(None = 자동) — 아는 측점 열쇠만."""
    root, _ = await e02._project(project_id)
    known = {item["key"] for item in plan_run(root, _ground(root))["items"]}
    unknown = sorted(set(payload.overrides) - known)
    if unknown:
        raise HTTPException(status_code=422, detail=f"모르는 측점: {unknown[:5]}")
    overrides = {k: v for k, v in sorted(payload.overrides.items()) if v is not None}
    atomic_write_json(root / REPRESENTATIVE, {"overrides": overrides})
    return {"overrides": overrides}


def _terrain(dtm: str, origin: list[float], strips: list[np.ndarray]) -> dict[str, Any] | None:
    """설계 지형 격자(E 복사본 DTM) — 노선면 둘레 띠만 · 노면 발자국 안 = None(깎기 지형이 노면을
    덮지 않게 · 가장자리 틈 한 칸 허용). 꼴 = E04 Grid + z(행 우선 iy·nx+ix) · 리본과 같은 원점."""
    try:
        sampler = DtmGridSampler.open_cached(dtm)
    except (OSError, ValueError, KeyError):
        logger.warning("E03 3D 지형 격자 못 엶: %s", dtm)
        return None
    pts = np.vstack([p.reshape(-1, 3)[:, :2] for p in strips])
    lo = pts.min(axis=0) - TERRAIN_MARGIN_M
    hi = pts.max(axis=0) + TERRAIN_MARGIN_M
    area = float(np.prod(hi - lo))
    dx = max(TERRAIN_STEP_M, math.ceil(math.sqrt(area / TERRAIN_MAX_CELLS) * 10) / 10)
    nx, ny = (np.floor((hi - lo) / dx).astype(int) + 1).tolist()
    xx, yy = np.meshgrid(lo[0] + np.arange(nx) * dx, lo[1] + np.arange(ny) * dx)
    flat = np.column_stack([xx.ravel(), yy.ravel()])
    near = np.isfinite(cKDTree(pts).query(flat, distance_upper_bound=TERRAIN_MARGIN_M)[0])
    # 발자국 = 리본마다 첫 열 + 끝 열(거꾸로) 다각형의 합집합
    rings = [np.vstack([p[:, 0, :2], p[::-1, -1, :2]]) for p in strips if len(p) >= 2]
    foot = shapely.union_all([shapely.make_valid(shapely.Polygon(r)) for r in rings])
    hole = shapely.contains_xy(foot, flat[:, 0], flat[:, 1])
    z, valid = sampler.sample_xy(flat + np.asarray(origin[:2]))
    keep = near & valid & ~hole & np.isfinite(z)
    local = np.round(z - origin[2], 2)
    return {
        "grid": {
            "x0": round(float(lo[0]), 3),
            "y0": round(float(lo[1]), 3),
            "dx": dx,
            "nx": nx,
            "ny": ny,
        },
        "z": [float(v) if k else None for v, k in zip(local.tolist(), keep.tolist())],
    }


@lru_cache(maxsize=2)
def _corridor3d(
    path: str, _stamp: tuple[int, int], dtm: str | None, _dtm_stamp: tuple[int, int] | None
) -> dict[str, Any]:
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    ribbons = []
    strips = []
    origin = None
    for ribbon in data.get("ribbons") or []:
        cols = int(ribbon["colCount"])
        chain = np.asarray(ribbon["chainages"], dtype=float)
        pos = np.frombuffer(base64.b64decode(ribbon["positionsBase64"]), dtype="<f8")
        pos = pos.reshape(len(chain), cols, 3)
        if origin is None and len(pos):
            origin = pos[0, 0].tolist()
        # 솎음 — 1 m 마다 첫 줄 + 끝 줄
        bins = np.floor(np.round(chain, 6) / CORRIDOR_STEP_M)
        keep = np.flatnonzero(np.r_[True, np.diff(bins) > 0])
        if keep[-1] != len(chain) - 1:
            keep = np.r_[keep, len(chain) - 1]
        local = pos[keep] - np.asarray(origin)
        strips.append(local)
        ribbons.append(
            {
                "kind": ribbon["kind"],
                "side": ribbon["side"],
                "colCount": cols,
                "chainages": np.round(chain[keep], 3).tolist(),
                "positions": np.round(local, 3).reshape(-1).tolist(),
            }
        )
    terrain = _terrain(dtm, origin, strips) if dtm and origin and strips else None
    return {"origin": origin, "ribbons": ribbons, "terrain": terrain}


@router.get("/{project_id}/stab/corridor3d", response_model=None)
async def corridor3d(project_id: UUID) -> dict[str, Any]:
    """복사본 노선면 — 원점 빼고 1 m 솎음(모델 x · y · z) + 설계 지형 격자(노면 구멍)."""
    root, _ = await e02._project(project_id)
    source = src.source_dir(root)
    path = src.corridor_path(source)
    if not path:
        raise HTTPException(status_code=404, detail="복사본에 노선면이 없습니다.")
    stat = path.stat()
    try:
        dtm = src.terrain_grid_path(source)
        dtm_stat = dtm.stat()
        dtm_key = (str(dtm), (dtm_stat.st_mtime_ns, dtm_stat.st_size))
    except (OSError, FileNotFoundError):
        dtm_key = (None, None)
    return await asyncio.to_thread(
        _corridor3d, str(path), (stat.st_mtime_ns, stat.st_size), *dtm_key
    )