"""532-3 E03 2D 측점 해석 길 · 화면 계산(요구 345 · 347 · 351).

- 길: 지문이 같으면 다시 안 셈 · 바뀐 측점만 다시 셈 · 진행률 · 결과 꼴 · 대표 단면 저장 ·
  공유 회사 POST 403(공용 접근 판정) · 노선면 솎음
- 실제 엔진 한 측점 → 화면 꼴(임계면 · 절편 · 격자)
- logic(TS) — 판정 규칙 = 파이썬 · 정렬 · 거름 · 등안전율 띠 · 화면 좌표 · 평면 띠 · 힘 화살
입력 = 검증_sub2 원본을 임시 폴더로 동결 복사(DB 읽기만) · 엔진 = 가짜(길) / 진짜(꼴).
"""

import asyncio
import copy
import json
import math
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
from uuid import UUID

import aiomysql
import pytest
from fastapi import HTTPException

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

from common_util import common_util_auth as auth  # noqa: E402
from common_util import common_util_project_copy as cp  # noqa: E402
from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER  # noqa: E402
from E01_StabRegister import E01_StabRegister_Source as src  # noqa: E402
from E02_StabGround import E02_StabGround_Model as ground_model  # noqa: E402
from E03_StabSection import E03_StabSection_Router as router  # noqa: E402
from E03_StabSection import E03_StabSection_View as view  # noqa: E402
from E03_StabSection import E03_StabSection_Worker as worker  # noqa: E402
from E03_StabSection.E03_StabSection_Engine_Model import section_model_from_b06  # noqa: E402
from E03_StabSection.E03_StabSection_Engine_Station import analyze_station  # noqa: E402

SOURCE_ID = "cb15892a-5788-4a26-9316-20a15421f68b"  # 검증_sub2(읽기만)
PID = "e" * 8 + "-" + "e" * 4 + "-" + "e" * 4 + "-" + "e" * 4 + "-" + "e" * 12
N_STATIONS = 3
TYPESCRIPT = ROOT / "config" / "node_modules" / "typescript"
TS = ROOT / "E03_StabSection" / "logic" / "E03_StabSection_Logic.ts"


async def _copy(target: Path) -> None:
    connection = await aiomysql.connect(
        host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, db=DB_NAME, autocommit=True
    )
    try:
        await cp.copy_project(
            connection, SOURCE_ID, scope="stability", new_id=PID, project_root=target
        )
    finally:
        connection.close()


@pytest.fixture(scope="module")
def frozen():
    folder = Path(tempfile.mkdtemp(prefix="aislo_e03_"))
    try:
        asyncio.run(_copy(folder / "copy"))
    except Exception as error:  # noqa: BLE001
        shutil.rmtree(folder, ignore_errors=True)
        pytest.skip(f"검증 원본 복사 실패: {error}")
    yield folder / "copy"
    shutil.rmtree(folder, ignore_errors=True)


CALLS: list[float] = []


def _fake_analyze(cross, design, *, layers=None, ground=None, conditions=("dry", "wet"), **kw):
    """엔진 출력 꼴 그대로 · 값은 누가거리로 정함(1.0 + ch/100 · 건기 +0.3)."""
    CALLS.append(float(cross["chainage_m"]))
    model = section_model_from_b06(
        cross, design, layers=layers, ground=ground, materials=kw.get("materials")
    )
    cases = []
    for cond in conditions:
        for side in model["sides"]:
            F = round(1.0 + float(cross["chainage_m"]) / 100 + (0.3 if cond == "dry" else 0.0), 3)
            x0, x1 = side["limits"]
            row = {
                "x_left": x0,
                "x_right": x1,
                "width": x1 - x0,
                "height": 1.0,
                "alpha_deg": 30.0,
                "weight": 20.0,
                "surcharge": 0.0,
                "pore_pressure": 0.0,
                "base_length": 2.0,
                "material": "weathered_soil",
                "c": 15.0,
                "phi_deg": 30.0,
                "normal_eff": 15.0,
                "shear_mobilized": 9.0,
                "E_left": 0.0,
                "X_left": 0.0,
            }
            found = {
                "F": F,
                "kind": "circle",
                "points": [[x0, 800.0], [x1, 799.0]],
                "entry": [x0, 800.0],
                "exit": [x1, 799.0],
                "iterations": 4,
                "converged": True,
                "slices": [row],
                "center": [0.0, 820.0],
                "radius": 20.0,
            }
            result = {
                "direction": side["direction"],
                "limits": list(side["limits"]),
                "methods": {"bishop": {"circular": found, "min": found}},
                "contour": {
                    "method": "bishop",
                    "x": [0.0, 1.0],
                    "z": [810.0, 811.0],
                    "F": [[F, None], [2.0, 3.0]],
                },
                "counts": {"circles": 1},
            }
            cases.append({"condition": cond, **side, "min_F": {"bishop": F}, "result": result})
    return {
        "chainage_m": cross["chainage_m"],
        "label": cross.get("label"),
        "cases": cases,
        "areas": model["areas"],
        "elapsed_s": 0.0,
    }


@pytest.fixture
def project(frozen, monkeypatch):
    """E 프로젝트 루트(복사본 + 비운 결과) · 측점 셋 · 가짜 엔진 · 서버 안에서 차례로."""
    for name in ("_derived", "E02_StabGround", "E03_StabSection", "initial_snapshot"):
        shutil.rmtree(frozen / name, ignore_errors=True)
    original = src.stations
    monkeypatch.setattr(router.src, "stations", lambda root: original(root)[:N_STATIONS])

    async def fake_project(_pid):
        return frozen, {"company_id": 1, "project_kind": "STABILITY"}

    monkeypatch.setattr(router.e02, "_project", fake_project)
    monkeypatch.setattr(worker, "analyze_station", _fake_analyze)
    monkeypatch.setattr(router, "WORKERS", 1)
    router._JOBS.clear()
    CALLS.clear()
    return frozen


def _analyze() -> dict:
    """POST → 뒤 일이 끝날 때까지 기다림."""

    async def go():
        first = await router.analyze2d(UUID(PID))
        while router._TASKS:
            await asyncio.gather(*list(router._TASKS))
        return first

    return asyncio.run(go())


USER = {"role": "USER", "company_id": 1}


def test_지문이_같으면_다시_안셈(project):
    first = _analyze()
    assert first["status"] == "accepted" and len(CALLS) == N_STATIONS
    job = asyncio.run(router.progress(UUID(PID)))
    assert job == {**job, "running": False, "percent": 100, "error": None, "computed": N_STATIONS}

    again = _analyze()
    assert again == {"status": "fresh", "fingerprint": first["fingerprint"]}
    assert len(CALLS) == N_STATIONS  # 다시 안 셈

    # 둘째 측점만 구간 덮어쓰기 → 그 측점만 다시 셈 · 다른 지문
    ground = router._ground(project)
    second = router.plan_run(project, ground)["items"][1]["chainage_m"]
    ground["zones"] = [
        {
            "from_m": second,
            "to_m": second,
            "soil": "colluvium",
            "rock": None,
            "soil_thickness_m": None,
        }
    ]
    (project / "E02_StabGround").mkdir(exist_ok=True)
    ground_model.ground_path(project).write_text(json.dumps(ground), encoding="utf-8")
    CALLS.clear()
    third = _analyze()
    assert third["status"] == "accepted" and third["fingerprint"] != first["fingerprint"]
    assert CALLS == [second]

    # 기준안전율만 바꿈 → 엔진은 안 돌고 판정만 새로(요약 지문만 바뀜)
    ground["safety_factors"]["dry"]["fill"] = 9.0
    ground_model.ground_path(project).write_text(json.dumps(ground), encoding="utf-8")
    CALLS.clear()
    assert _analyze()["status"] == "accepted" and CALLS == []
    body = asyncio.run(router.result2d(UUID(PID), USER))
    fill_dry = [
        c["dry"]["left"] for c in (r["cases"] for r in body["stations"]) if "left" in c["dry"]
    ]
    assert body["fresh"] and any(v["required"] == 9.0 and v["ok"] is False for v in fill_dry)


def test_진행률_올라감(project, monkeypatch):
    seen = []

    def spy(*args, **kw):
        seen.append(router._JOBS[PID]["percent"])
        return _fake_analyze(*args, **kw)

    monkeypatch.setattr(worker, "analyze_station", spy)
    _analyze()
    assert seen[0] == 2 and seen == sorted(seen) and len(set(seen)) == N_STATIONS
    assert asyncio.run(router.progress(UUID(PID)))["percent"] == 100


def test_결과_꼴_대표단면(project):
    none = asyncio.run(router.result2d(UUID(PID), USER))
    assert none["status"] == "none" and len(none["stations"]) == N_STATIONS
    _analyze()
    body = asyncio.run(router.result2d(UUID(PID), USER))
    assert body["status"] == "ok" and body["fresh"] and body["editable"]
    assert [r["key"] for r in body["stations"]] == [r["key"] for r in none["stations"]]
    row = body["stations"][0]
    assert set(row["verdict"]) == {"dry", "wet"} and row["representative"]["on"] is True  # 하위 5
    other = asyncio.run(router.result2d(UUID(PID), {"role": "USER", "company_id": 2}))
    assert other["editable"] is False

    sec = asyncio.run(router.section(UUID(PID), row["key"]))
    assert sec["regions"] and sec["surface"] and sec["surcharge"][0]["q"] == 10.0
    assert sec["water"]["dry"] is None and sec["water"]["wet"]
    a = sec["analyses"][0]
    assert a["critical"]["points"] and a["grid"]["fs"][0][1] is None and a["slices"][0]["N"] == 15.0
    with pytest.raises(HTTPException) as missing:
        asyncio.run(router.section(UUID(PID), "99999.00"))
    assert missing.value.status_code == 404

    saved = asyncio.run(
        router.put_representative(
            UUID(PID), router.RepresentativeRequest(overrides={row["key"]: False})
        )
    )
    assert saved == {"overrides": {row["key"]: False}}
    after = asyncio.run(router.result2d(UUID(PID), USER))["stations"][0]["representative"]
    assert after == {"auto": ["lowest"], "user": False, "on": False}
    with pytest.raises(HTTPException) as bad:
        asyncio.run(
            router.put_representative(
                UUID(PID), router.RepresentativeRequest(overrides={"x": True})
            )
        )
    assert bad.value.status_code == 422


def test_노선면_솎음(project):
    body = asyncio.run(router.corridor3d(UUID(PID)))
    assert len(body["origin"]) == 3 and body["ribbons"]
    for ribbon in body["ribbons"]:
        chain = ribbon["chainages"]
        assert len(ribbon["positions"]) == len(chain) * ribbon["colCount"] * 3
        assert chain == sorted(set(chain))  # 1 m 칸마다 한 줄 + 끝 줄
        assert len(chain) <= math.ceil((chain[-1] - chain[0]) / router.CORRIDOR_STEP_M) + 2
        assert max(abs(v) for v in ribbon["positions"]) < 5000  # 원점 뺀 자리


class _Cursor:
    def __init__(self, rows):
        self.rows = rows

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def execute(self, sql, args=()):
        self.last = sql

    async def fetchone(self):
        return self.rows["shared"] if "plan_route_assignees" in self.last else self.rows["project"]


class _Pool:
    def __init__(self, rows):
        self.rows = rows

    def acquire(self):
        pool = self

        class _Conn:
            async def __aenter__(self):
                return self

            async def __aexit__(self, *exc):
                return False

            def cursor(self):
                return _Cursor(pool.rows)

        return _Conn()


@pytest.mark.parametrize("method,ok", [("GET", True), ("POST", False), ("PUT", False)])
def test_공유_회사는_GET_만(monkeypatch, method, ok):
    monkeypatch.setattr(auth, "get_db_pool", lambda: _Pool({"project": (1,), "shared": (1,)}))
    request = SimpleNamespace(path_params={"project_id": PID}, method=method)
    session = {"role": "USER", "company_id": 2}
    if ok:
        assert asyncio.run(auth.require_project_access(request, session)) is session
    else:
        with pytest.raises(HTTPException) as error:
            asyncio.run(auth.require_project_access(request, session))
        assert error.value.status_code == 403


def test_E03_길에_공용_접근_판정이_걸림():
    import main

    paths = {
        "/api/projects/{project_id}/stab/" + p
        for p in ("analyze2d", "result2d", "progress", "representative")
    }
    found = set()
    for route in main.app.routes:
        if getattr(route, "path", None) in paths:
            calls = {d.call for d in route.dependant.dependencies}
            assert auth.require_project_access in calls and auth.require_company in calls
            found.add(route.path)
    assert found == paths


def test_실제_엔진_한_측점_화면_꼴(frozen):
    source = src.source_dir(frozen)
    station = src.stations(source)[5]
    cross = src.cross_section(source, station["chainage_m"])
    out = analyze_station(cross, cross["design"], methods=("ordinary", "bishop"), noncircular=False)
    meta = {
        "key": "k",
        "chainage_m": station["chainage_m"],
        "label": cross["label"],
        "section_mode": cross["design"]["section_mode"],
        "center": [0, 0],
        "fp": "f",
    }
    factors = {"dry": {"cut": 1.5, "fill": 1.5}, "wet": {"cut": 1.2, "fill": 1.3}}
    row = view.station_row(meta, out, factors)
    model = section_model_from_b06(cross, cross["design"])
    sec = view.section_view(row, out, model, {"dry": None, "wet": model["surface"]})
    assert {a["condition"] for a in sec["analyses"]} == {"dry", "wet"}
    for a in sec["analyses"]:
        c = a["critical"]
        assert c and c["kind"] == "circle" and c["center"] and c["radius"] > 0
        # 임계면 꼭짓점 = 원 위(반경 ±0.01 m)
        for x, z in c["points"]:
            assert abs(math.hypot(x - c["center"][0], z - c["center"][1]) - c["radius"]) < 0.02
        assert a["slices"] and a["slices"][0]["x_left"] == pytest.approx(
            min(p[0] for p in c["points"]), abs=0.01
        )
        assert a["min_by_method"]["bishop"] == pytest.approx(c["fs"], abs=1e-3)
        assert row["cases"][a["condition"]][a["side"]]["ok"] == (c["fs"] >= a["required"])
        assert len(a["grid"]["fs"]) == len(a["grid"]["z"])


def test_대표단면_자동_규칙():
    rows = [
        {"key": str(i), "sides": [{"role": "fill", "height_m": h}], "min": {"dry": 3.0 - i * 0.1}}
        for i, h in enumerate([10.0, 9.99, 0, 0, 0, 0, 0, 0])
    ]
    rows[3]["sides"] = [{"role": "cut", "height_m": 15.0}]
    auto = view.auto_representative(rows)
    assert (
        auto["0"] == ["fill_height"] and "1" not in auto and auto["3"] == ["cut_height", "lowest"]
    )
    assert sorted(k for k, v in auto.items() if "lowest" in v) == ["3", "4", "5", "6", "7"]


# ---- logic(TS) --------------------------------------------------------------
SCRIPT = r"""
const ts = require(process.argv[1]);
const fs = require("fs");
const src = fs.readFileSync(process.argv[2], "utf8");
const opts = { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 };
const js = ts.transpileModule(src, { compilerOptions: opts }).outputText;
const box = { exports: {} };
new Function("exports", "module", js)(box.exports, box);
const m = box.exports;
const input = JSON.parse(fs.readFileSync(0, "utf8"));
const fit = m.fitView(input.view, 800, 500, 20);
const shape = input.shape;
process.stdout.write(JSON.stringify({
  verdict: input.rows.map((r) => [m.verdictOf(r), m.verdictOf(r, "dry"), m.verdictOf(r, "wet")]),
  sortFs: m.listRows(input.rows, "wet", "fs", false).map((r) => r.key),
  ngOnly: m.listRows(input.rows, "wet", "chainage", true).map((r) => r.key),
  bands: input.fsList.map((f) => m.bandIndex(f, 1.5)),
  legend: m.bandLegend(1.5),
  fit,
  matrix: m.viewMatrix(fit),
  screen: input.view.surface.map((p) => m.toScreen(fit, p)),
  back: input.view.surface.map((p) => m.toModel(fit, m.toScreen(fit, p))),
  segments: m.bandSegments(input.rows, "wet").map((s) => [s.verdict, s.keys]),
  arrows: m.forceArrows(input.slice, shape, -1, 0.01),
  arrowsPos: m.forceArrows(input.slice, shape, 1, 0.01),
  stress: m.sliceStress([input.slice]),
  ticks: m.niceTicks(-3, 47, 5),
  zAt: [m.zAt([[0, 0], [10, 10]], 2.5), m.zAt([[10, 10], [0, 0]], -5)],
}));
"""


def _row(key, ch, dry, wet, center):
    cases = {
        "dry": {"left": {"fs": dry, "required": 1.5, "ok": None if dry is None else dry >= 1.5}},
        "wet": {"left": {"fs": wet, "required": 1.3, "ok": None if wet is None else wet >= 1.3}},
    }
    verdict = {}
    for cond, by_side in cases.items():
        oks = [v["ok"] for v in by_side.values() if v["ok"] is not None]
        verdict[cond] = "none" if not oks else ("ok" if all(oks) else "ng")
    return {
        "key": key,
        "chainage_m": ch,
        "label": key,
        "center": center,
        "cases": cases,
        "sides": [{"side": "left", "role": "fill", "height_m": 1.0}],
        "min": {"dry": dry, "wet": wet},
        "verdict": verdict,
        "error": None,
    }


def test_logic_TS():
    rows = [
        _row("a", 0, 2.0, 1.4, [0, 0]),
        _row("b", 20, 1.6, 1.1, [20, 0]),
        _row("c", 40, None, None, [40, 5]),
        _row("d", 60, 1.2, 1.0, [60, 10]),
        _row("e", 80, 2.0, 1.5, [80, 10]),
    ]
    surface = [[-20.0, 810.0], [-2.0, 800.0], [2.0, 800.0], [20.0, 790.0]]
    sec = {"surface": surface, "bottom_z": 780.0, "analyses": []}
    shape = [[0.0, 0.0], [2.0, 1.0], [2.0, 4.0], [0.0, 3.0]]
    slice_row = {
        "x_left": 0,
        "x_right": 2,
        "weight": 50,
        "surcharge": 5,
        "N_eff": 40,
        "U": 10,
        "S": 20,
        "E_left": 3,
        "E_right": 4,
        "X_left": 0,
        "X_right": 0,
        "base_length": 2.236,
        "pore_pressure": 4.47,
    }
    fs_list = [None, 1.0, 1.2, 1.35, 1.5, 1.65, 1.875, 2.25, 9.0]
    payload = {"rows": rows, "view": sec, "fsList": fs_list, "shape": shape, "slice": slice_row}
    proc = subprocess.run(
        ["node", "-e", SCRIPT, str(TYPESCRIPT), str(TS)],
        input=json.dumps(payload),
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=60,
    )
    assert proc.returncode == 0, proc.stderr
    out = json.loads(proc.stdout)

    # 판정 규칙 = 파이썬 overall_verdict
    for r, got in zip(rows, out["verdict"]):
        assert got == [
            view.overall_verdict(r),
            view.overall_verdict(r, "dry"),
            view.overall_verdict(r, "wet"),
        ]
    assert out["sortFs"] == ["d", "b", "a", "e", "c"]  # 없음은 맨 뒤
    assert out["ngOnly"] == ["b", "d"]
    # 띠 = 안전율 ÷ 기준(1.5) 경계 0.8 · 0.9 · 1.0 · 1.1 · 1.25 · 1.5
    assert out["bands"] == [None, 0, 1, 2, 3, 4, 5, 6, 6]
    assert [(b["from"], b["to"]) for b in out["legend"]] == [
        (None, 1.2),
        (1.2, 1.35),
        (1.35, 1.5),
        (1.5, 1.65),
        (1.65, 1.88),
        (1.88, 2.25),
        (2.25, None),
    ]
    # 화면 좌표 — 좌측(+offset) 이 왼쪽 · 위 = 높은 표고 · 같은 축척 · 왕복 = 원래 값 · 판 안
    k, tx, ty = out["fit"]["k"], out["fit"]["tx"], out["fit"]["ty"]
    assert out["matrix"] == f"matrix({-k} 0 0 {-k} {tx} {ty})"
    xs = [p[0] for p in out["screen"]]
    assert xs == sorted(xs, reverse=True)
    for (x, z), (sx, sy), back in zip(surface, out["screen"], out["back"]):
        assert (sx, sy) == pytest.approx((-k * x + tx, -k * z + ty))
        assert back == pytest.approx([x, z]) and 0 <= sx <= 800 and 0 <= sy <= 500
    # 평면 띠 — 판정이 바뀔 때만 끊음 · 색 수 = 판정 수
    assert out["segments"] == [
        ["ok", ["a"]],
        ["ng", ["b"]],
        ["none", ["c"]],
        ["ng", ["d"]],
        ["ok", ["e"]],
    ]
    assert len({s[0] for s in out["segments"]}) == len(
        {view.overall_verdict(r, "wet") for r in rows}
    )

    def vec(a):
        return (a["to"][0] - a["from"][0], a["to"][1] - a["from"][1])

    arrows = {a["name"]: a for a in out["arrows"]}
    assert set(arrows) == {"W", "Q", "N", "U", "S", "EL", "ER"}  # X = 0 은 뺌
    assert vec(arrows["W"]) == pytest.approx((0, -0.5))  # 50 × 0.01 아래
    base = (2 / math.sqrt(5), 1 / math.sqrt(5))
    nx, nz = vec(arrows["N"])
    assert nx * base[0] + nz * base[1] == pytest.approx(0, abs=1e-9) and nz > 0  # 바닥에 수직 · 위
    sx, sz = vec(arrows["S"])
    assert (sx, sz) == pytest.approx(
        (0.2 * base[0], 0.2 * base[1])
    )  # 방향 −1 → 미끄럼 반대 = +바닥
    assert vec({a["name"]: a for a in out["arrowsPos"]}["S"]) == pytest.approx(
        (-0.2 * base[0], -0.2 * base[1])
    )
    assert arrows["EL"]["to"] == pytest.approx(
        [2.0, 2.5]
    )  # 방향 −1 → 틀의 왼쪽 면 = 결과 오른쪽 면
    assert vec(arrows["EL"]) == pytest.approx((-0.03, 0))
    assert out["stress"][0] == pytest.approx(
        {"x": 1, "sigma": 40 / 2.236, "u": 4.47, "tau": 20 / 2.236}
    )
    assert out["ticks"][0] <= -3 and out["ticks"][-1] >= 47 and 0 in out["ticks"]
    assert out["zAt"] == [2.5, 0]


def test_엔진_지문은_코드_내용():
    assert len(router.engine_hash()) == 12
    item = copy.deepcopy({"upper": "a", "lower": "b", "soil_thickness_m": 0.5})
    assert router._layers(item) == [["a", 0.5], ["b", None]]
    assert router._layers({"upper": "a", "lower": None}) == [["a", None]]
    assert router._layers(None) is None