"""E02 [해석 시작] 체인 (요구 390 1 · PLAN 564-1).

차례 = 2D → 3D(모든 구간 · 건기 · 우기) → 암반 → 광역 → 보고서. 계산 자리 = ③ 서버 단독.
각 페이지 해석 함수를 그대로 부름(그 파일 안 고침) · 단계마다 입력 지문이 저장본과 같으면 건너뜀
· 결과는 각 페이지 저장 자리에 영구 저장 · 한 단계가 실패해도 다음 단계로.
도는 동안 그 페이지 진행 표(`_JOBS`)에 같은 일감을 꽂음 → E03 · E04 · E07 화면이 「해석 중」
· 그 페이지 [해석] 은 409. [끊기] = 다음 확인 때 멈춤(3D 는 엔진 안 · 그 밖은 단계 사이).

진행 꼴 = {running, percent, stage, stage_name, steps:[{key, name, state, seconds}],
  message, error, cancelled} · state = wait · run · done · skip · error · cancel
"""

from __future__ import annotations

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any
from uuid import UUID

from common_util.common_util_workflow_state import mark_stab_step

logger = logging.getLogger(__name__)

#: 단계 — (열쇠, 이름, 비율 몫) · 몫 = 1.2 km 실측 시간 비(2D 1~3 분 · 3D 건기 · 우기 약 25 분)
STEPS: tuple[tuple[str, str, float], ...] = (
    ("2d", "2D 측점 해석", 10.0),
    ("3d", "3D 구간 해석", 84.0),
    ("rock", "암반 해석", 2.0),
    ("area", "광역 위험", 2.0),
    ("report", "보고서", 2.0),
)
CONDITIONS_3D = ("dry", "wet")

Report = Callable[[float, str], None]  # (단계 안 비율 0~1, 글)
Stopped = Callable[[], bool]
Step = Callable[[Report, Stopped], Awaitable[str]]  # → "done" | "skip"

_JOBS: dict[str, dict[str, Any]] = {}
_STOPS: dict[str, bool] = {}
_TASKS: set[asyncio.Task] = set()


def idle() -> dict[str, Any]:
    return {
        "running": False,
        "percent": 0,
        "stage": None,
        "stage_name": None,
        "steps": [{"key": k, "name": n, "state": "wait", "seconds": None} for k, n, _ in STEPS],
        "message": None,
        "error": None,
        "cancelled": False,
    }


async def _mark_stage(project_id: str, key: str, state: str, message: str | None = None) -> None:
    """대시보드 단계 상태(565-1) — 적기 실패는 체인을 막지 않음."""
    if not project_id:
        return
    try:
        await mark_stab_step(project_id, key, state, message)
    except Exception:  # noqa: BLE001
        logger.exception("E02 체인 단계 상태 적기 실패: %s %s", key, state)


async def run_chain(
    job: dict[str, Any], steps: dict[str, Step], stopped: Stopped, project_id: str = ""
) -> None:
    """STEPS 차례로 돎 — job 을 제자리에서 고침(진행 GET 이 그대로 읽음)."""
    total = sum(w for _, _, w in STEPS)
    base = 0.0
    errors: list[str] = []
    for index, (key, name, weight) in enumerate(STEPS):
        row = job["steps"][index]
        if stopped():
            job.update(cancelled=True, message="끊음")
            row["state"] = "cancel"
            await _mark_stage(project_id, key, "cancel")
            break
        job.update(stage=key, stage_name=name, message=name)
        row["state"] = "run"
        await _mark_stage(project_id, key, "run")

        def report(frac: float, message: str, base: float = base, weight: float = weight) -> None:
            frac = min(1.0, max(0.0, float(frac)))
            job.update(percent=min(99, int(100 * (base + weight * frac) / total)), message=message)

        started = time.perf_counter()
        try:
            row["state"] = await steps[key](report, stopped)
        except InterruptedError:
            row["state"] = "cancel"
            job.update(cancelled=True, message="끊음")
            break
        except Exception as error:  # noqa: BLE001 — 단계 실패 격리 · 까닭은 그대로 올림
            logger.exception("E02 체인 %s 실패: project_id=%s", key, project_id)
            row["state"] = "error"
            errors.append(f"{name}: {getattr(error, 'detail', None) or error}")
        finally:
            row["seconds"] = round(time.perf_counter() - started, 1)
            failed = errors[-1] if row["state"] == "error" else None
            await _mark_stage(project_id, key, row["state"], failed)
        base += weight
        job["percent"] = min(99, int(100 * base / total))
    job["error"] = " · ".join(errors) or None
    if not job["cancelled"]:
        job.update(percent=100, message="끝")
    job.update(stage=None, stage_name=None)


# ---------------------------------------------------------------------------
# 실제 단계 — 페이지 함수 부르기(늦은 import · E03 · E06 라우터가 E02 라우터를 import 함)
# ---------------------------------------------------------------------------


def _plug(jobs: dict[str, dict[str, Any]], pid: str, extra: dict[str, Any]) -> dict[str, Any]:
    """그 페이지 진행 표에 체인 일감을 꽂음 — 페이지 화면 「해석 중」 · 페이지 [해석] 409."""
    job = {"running": True, "percent": 0, "message": "한 번에 해석", "error": None, **extra}
    jobs[pid] = job
    return job


def build_steps(project_id: UUID, project_root: Path) -> dict[str, Step]:
    from E01_StabRegister import E01_StabRegister_Source as source
    from E03_StabSection import E03_StabSection_Router as e03
    from E04_Stab3D import E04_Stab3D_Router as e04
    from E04_Stab3D import E04_Stab3D_Run as run3d
    from E05_StabReport import E05_StabReport_Build as e05_build
    from E05_StabReport import E05_StabReport_Router as e05
    from E05_StabReport import E05_StabReport_Store as e05_store
    from E06_StabRock import E06_StabRock_Router as e06
    from E07_StabArea import E07_StabArea_Router as e07

    pid = str(project_id)
    root = project_root

    async def step_2d(report: Report, _stopped: Stopped) -> str:
        plan = await asyncio.to_thread(e03.plan_run, root, e03._ground(root))
        if e03.read_derived(root, e03.KIND_SUMMARY, plan["fp"]):
            return "skip"
        page = _plug(e03._JOBS, pid, {"computed": 0})

        def relay(percent: int, message: str) -> None:
            page.update(percent=int(percent), message=message)
            report(percent / 100, f"2D {message}")

        try:
            page["computed"] = await asyncio.to_thread(e03.run_all, root, plan, relay)
            page.update(percent=100, message="끝")
        finally:
            page["running"] = False
        return "done"

    def search_settings() -> dict[str, Any]:
        """3D 탐색 설정 = 가장 최근 저장본 설정 · 없으면 기본값(구간 · 조건 빼고)."""
        files = sorted(run3d.results_dir(root).glob("*.json"), key=lambda p: p.stat().st_mtime_ns)
        saved = run3d.load(root, files[-1].stem) if files else None
        fields = e04.SearchSettings.model_fields
        values = {k: v for k, v in ((saved or {}).get("settings") or {}).items() if k in fields}
        return e04.SearchSettings(**values).model_dump()

    async def step_3d(report: Report, stopped: Stopped) -> str:
        src = e04._source_root(root)
        stations = await asyncio.to_thread(source.stations, src)
        sections = run3d.auto_sections(stations)
        search = await asyncio.to_thread(search_settings)
        units = [(s, c) for s in sections for c in CONDITIONS_3D]
        computed = 0
        for index, (section, condition) in enumerate(units):
            if stopped():
                raise InterruptedError
            settings = {
                "start_m": round(section["start_m"], 2),
                "end_m": round(section["end_m"], 2),
                "condition": condition,
                **search,
            }
            key = run3d.section_key(settings["start_m"], settings["end_m"], condition)
            label = f"3D {index + 1} / {len(units)} 구간"
            finger = await asyncio.to_thread(e04._current_fingerprint, root, src, settings)
            saved = await asyncio.to_thread(run3d.load, root, key)
            if saved and saved["fingerprint"] == finger:
                report((index + 1) / len(units), label)
                continue
            page = _plug(e04._JOBS, pid, {**e04._IDLE, "running": True, "key": key})
            e04._STOPS[pid] = False

            def relay(percent: int, message: str, index: int = index, page=page) -> None:
                page.update(percent=max(0, min(99, int(percent))), message=message)
                report((index + percent / 100) / len(units), label)

            try:
                page["summary"] = await asyncio.to_thread(
                    run3d.analyze,
                    root,
                    src,
                    settings,
                    finger,
                    relay,
                    lambda: stopped() or e04._STOPS.get(pid, False),
                )
                page.update(percent=100, message="끝")
            except InterruptedError:
                page.update(cancelled=True, message="끊음")
                raise
            finally:
                page["running"] = False
            computed += 1
        return "done" if computed else "skip"

    async def step_rock(_report: Report, _stopped: Stopped) -> str:
        out = await asyncio.to_thread(e06._run, root)
        return "skip" if out["status"] == "fresh" else "done"

    async def step_area(report: Report, _stopped: Stopped) -> str:
        src = e07._source_root(root)
        saved = await asyncio.to_thread(e07.load, root)
        # E07 화면과 같음 — 지난 조건(없으면 기본값) + 물성은 지금 E02 값(564-4)
        base = {**e07.defaults(root), **((saved or {}).get("settings") or {})}
        mat = e07.ground_material(root, base.get("material_id"))
        if mat and mat["id"] == base.get("material_id"):
            base.update({k: float(mat[k]) for k in e07.MATERIAL_KEYS})
            base.update(c_hi=base["c"] + e07.RANGE_C_KPA, phi_hi=base["phi"] + e07.RANGE_PHI_DEG)
        settings = e07.AreaRequest(**base).model_dump(exclude={"force"})
        finger = e07.fingerprint(
            source.manifest(src)["source_fingerprint"],
            settings,
            e07.ground_material(root, settings["material_id"]),
        )
        if saved and saved["fingerprint"] == finger:
            return "skip"
        page = _plug(e07._JOBS, pid, {})

        def relay(percent: int, message: str) -> None:
            page.update(percent=max(0, min(99, int(percent))), message=message)
            report(percent / 100, f"광역 {message}")

        try:
            await asyncio.to_thread(e07.run, root, src, settings, finger, relay)
            page.update(percent=100, message="끝")
        finally:
            page["running"] = False
        return "done"

    async def step_report(_report: Report, _stopped: Stopped) -> str:
        async with e05._lock(project_id):
            book, meta = e05_store.load(root)
            fresh = await asyncio.to_thread(e05_build.analysis_fingerprint, root)
            if book is not None and meta and meta.get("fingerprint") == fresh:
                return "skip"
            await asyncio.to_thread(e05_store.refresh, root)
        return "done"

    return {
        "2d": step_2d,
        "3d": step_3d,
        "rock": step_rock,
        "area": step_area,
        "report": step_report,
    }


def busy(project_id: UUID) -> bool:
    """체인 · 페이지 해석 하나라도 도는지 — 한 프로젝트에 체인 하나만."""
    from E03_StabSection import E03_StabSection_Router as e03
    from E04_Stab3D import E04_Stab3D_Router as e04
    from E07_StabArea import E07_StabArea_Router as e07

    pid = str(project_id)
    return any(
        (jobs.get(pid) or {}).get("running") for jobs in (_JOBS, e03._JOBS, e04._JOBS, e07._JOBS)
    )


def start(project_id: UUID, project_root: Path) -> dict[str, Any]:
    """뒤에서 체인 시작 — 부르는 쪽이 busy 를 먼저 봄."""
    pid = str(project_id)
    job = {**idle(), "running": True, "message": "시작"}
    _JOBS[pid] = job
    _STOPS[pid] = False
    steps = build_steps(project_id, project_root)

    async def work() -> None:
        try:
            await _mark_stage(pid, "ground", "done")  # 지반 = 길에서 저장을 마친 뒤 시작
            await run_chain(job, steps, lambda: _STOPS.get(pid, False), pid)
        finally:
            job["running"] = False

    task = asyncio.create_task(work(), name=f"E02 chain {pid}")
    _TASKS.add(task)
    task.add_done_callback(_TASKS.discard)
    return job


def progress(project_id: UUID) -> dict[str, Any]:
    return dict(_JOBS.get(str(project_id)) or idle())


def cancel(project_id: UUID) -> bool:
    pid = str(project_id)
    running = bool((_JOBS.get(pid) or {}).get("running"))
    if running:
        _STOPS[pid] = True
    return running