"""PLAN 511-B B-5 (요구 320-4) — 노선 · 따라감 서버 몫 = 화면과 같은 TS(늘 떠 있는 Node).

못박는 것
  - 노선 저장 폴리라인 파일(선 csv · 노드 csv · 곡선표 + 편집표) — Node `route.polyline` 길 =
    파이썬 길 바이트 같음(거울 CASES · 위반 글 딱 반 R 12.25 · 12.75 · 겹친 점 · 짧은 칸 · 고정 ·
    고정 길이 · 알려진 차 `KNOWN_ARC` = 직각 원호 점 ±1) ·
    검증_sub3 저장 노선(노드 · 편집표) 다시 씀 = 저장된 곡선표 · 노드 그대로.
  - 저장 좌표 맞춤(F1) — Node `facility.follow` + fill = 파이썬 짝
    `_fill_structure_coordinates_py`(좌표 없음 · 측점 낳음 · 낡은 좌표 · 옛 좌표 지킴 · 자리 바뀜 ·
    site · 읽을 때 · 노선 밖).
  - 남은 측점 계획(F3) — Node `stations.plan` = 파이썬 `plan_stations`(322-3 CASES · 묶음마다).
  - 공유 DB 안 씀 · storage 읽기만(검증_sub3 · 없으면 건너뜀) · 쓰기는 tmp.

    ./venv/Scripts/python.exe -m pytest resources/tester/test_511_b5_route_node.py -q
"""

from __future__ import annotations

import asyncio
import json
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
for folder in (ROOT, HERE):
    if str(folder) not in sys.path:
        sys.path.insert(0, str(folder))

from B05_Profile import B05_Profile_Route_Polyline_Write as writer  # noqa: E402
from common_util import common_util_route_follow as follow  # noqa: E402

pytestmark = pytest.mark.skipif(
    not (ROOT / "config" / "node_modules" / "vite").is_dir(), reason="프론트엔드 의존성 미설치"
)

SUB3 = (
    ROOT / "storage" / "1" / "3" / "12953f5a-5ba1-42ad-a427-8ac857a58bc2" / "B05_Profile" / "route"
)
FIXTURE = HERE / "fixtures" / "golden_472"
NAMES = ("planned_route.csv", "planned_route_nodes.csv", "planned_route_curves.json")
# 알려진 차(브레인 승인 · 511-B B-5) — 교각 딱 5° 배수(합성 직각)면 원호 칸 수가 libm 끝자리로 갈림
# (파이썬 89.99999999999992° → 17 칸 · Node 90.0° → 18 칸) · 원호 점 ±1 · 실데이터 저장 노선 0 차
KNOWN_ARC = {
    "직각 한 번 · R 자동",
    "R 못박음 30m",
    "R 이 접선 자리보다 커서 눌림",
    "겹친 점(0.3m) 섞임 — 서버와 같이 버려야 함",
    "딱 반 R(위반 글 짝수 쪽)",
}


def _files(folder: Path, monkeypatch, node: bool, args: dict) -> tuple[dict, dict]:
    folder.mkdir(parents=True, exist_ok=True)
    with monkeypatch.context() as patch:
        if not node:
            patch.setattr(writer, "node_polyline", lambda *a: None)
        else:  # Node 못 돌리면 파이썬으로 새는 것을 막음
            made = writer.node_polyline
            patch.setattr(writer, "node_polyline", lambda *a: made(*a) or pytest.fail("Node 없음"))
        summary = writer._write_planned_polyline(folder / NAMES[0], simplify=False, **args)
    return summary, {name: (folder / name).read_bytes() for name in NAMES}


def _cases() -> list[tuple[str, dict]]:
    import test_route_polyline_browser_mirror as mirror

    out = [
        (
            name,
            {
                "points": points,
                "radius_m": mirror.MIN_RADIUS_M,
                "curve_flags": flags,
                "radii": radii,
            },
        )
        for name, points, flags, radii in mirror.CASES
    ]
    corner = [(0.0, 0.0), (100.0, 0.0), (100.0, 100.0), (200.0, 100.0)]
    out += [
        (
            "딱 반 R(위반 글 짝수 쪽)",
            {
                "points": corner,
                "radius_m": 15.0,
                "curve_flags": [True] * 4,
                "radii": [None, 12.25, 12.75, None],
            },
        ),
        (
            "겹친 점 · 짧은 칸 · 고정 · 고정 길이",
            {
                "points": [(0.0, 0.0), (80.0, 0.0), (80.3, 0.1), (120.0, 60.0), (200.0, 60.0)],
                "radius_m": 12.0,
                "curve_flags": [True, True, True, False],
                "radii": [None, 30.0],
                "locks": [None, "radius", "arc", "bad", None],
                "arcs": [None, None, 20.5, None],
            },
        ),
        (
            "점 둘",
            {
                "points": [(0.0, 0.0), (10.0, 0.0)],
                "radius_m": 12.0,
                "curve_flags": [True, True],
                "radii": [None, None],
            },
        ),
    ]
    return out


@pytest.mark.parametrize("name,args", _cases(), ids=[name for name, _ in _cases()])
def test_노선_폴리라인_Node_파이썬_바이트_같음(tmp_path, monkeypatch, name, args):
    (py_sum, py), (node_sum, node) = (
        _files(tmp_path / mode, monkeypatch, mode == "node", args) for mode in ("py", "node")
    )
    if name in KNOWN_ARC and node[NAMES[0]] != py[NAMES[0]]:
        # 알려진 차 — 선 csv 원호 점 수만 ±1 · 노드 · 곡선표 · 편집표는 같음
        assert abs(node_sum["vertices"] - py_sum["vertices"]) == 1, name
        node, py = ({k: v for k, v in f.items() if k != NAMES[0]} for f in (node, py))
        node_sum, py_sum = ({**s, "vertices": 0} for s in (node_sum, py_sum))
    assert (node_sum, node) == (py_sum, py), name


def test_딱_반_R_위반_글은_파이썬_꼴(tmp_path, monkeypatch):
    args = dict(_cases())["딱 반 R(위반 글 짝수 쪽)"]
    _, files = _files(tmp_path, monkeypatch, True, args)
    text = files[NAMES[2]].decode("utf-8")
    assert "최소곡선반지름 미달(12.2 < 15.0m)" in text and "(12.8 < 15.0m)" in text


@pytest.mark.skipif(not (SUB3 / "planned_route_nodes.csv").is_file(), reason="검증_sub3 노선 없음")
def test_검증_sub3_저장_노선_다시_씀_같음(tmp_path, monkeypatch):
    from common_util.common_util_route_geometry import read_planned_route_csv

    table = json.loads((SUB3 / NAMES[2]).read_text(encoding="utf-8"))
    nodes = [(v.x, v.y) for v in read_planned_route_csv(SUB3 / NAMES[1]).vertices]
    edits = table["edits"]
    args = {
        "points": nodes,
        "radius_m": table["min_radius_m"],
        "curve_flags": [e["curve"] for e in edits],
        "radii": [e["radius_m"] for e in edits],
        "locks": [e["lock"] for e in edits],
        "arcs": [e["arc_m"] for e in edits],
        "station_snap": table.get("station_snap"),
        "station_snap_step": table.get("station_snap_step"),
    }
    py = _files(tmp_path / "py", monkeypatch, False, args)
    node = _files(tmp_path / "node", monkeypatch, True, args)
    assert node == py
    again = json.loads(node[1][NAMES[2]])
    assert again["curves"] == table["curves"] and again["edits"] == edits
    assert node[1][NAMES[1]] == (SUB3 / NAMES[1]).read_bytes()


def _line() -> list[list[float]]:
    plan = json.loads((FIXTURE / "plan.json").read_text(encoding="utf-8"))
    return plan["planned"]


def _items(line: list[list[float]]) -> tuple[list[SimpleNamespace], dict]:
    on = follow.LineString(line)

    def at(m: float) -> tuple[float, float]:
        p = on.interpolate(m)
        return round(p.x, 4), round(p.y, 4)

    def s(sid, chainage, x=None, y=None, placement="point", **span):
        return SimpleNamespace(
            structure_id=sid,
            placement=placement,
            chainage_m=chainage,
            start_m=span.get("start_m"),
            end_m=span.get("end_m"),
            x=x,
            y=y,
        )

    hold = at(150.0)
    items = [
        s("bare", 120.0),
        s("holds", 150.0, *hold),
        s("stale", 210.0, *at(180.0)),
        s("kept", 260.0, 1.0, 2.0),
        s("moved", 300.0, 3.0, 4.0),
        s("site", 340.0, *at(341.3), placement="site"),
        s("span", 380.0, start_m=370.0, end_m=395.0),
        s("none", None, 5.0, 6.0),
        s("beyond", 99999.0),
        s("prev-bare", 420.0),
    ]
    previous = {
        "kept": s("kept", 260.0, 208000.5, 367800.25),
        "moved": s("moved", 290.0, 9.0, 9.0),
        "prev-bare": s("prev-bare", 420.0, 7.0, 8.0),
    }
    return items, previous


@pytest.mark.parametrize("only_missing", [False, True])
def test_좌표_맞춤_Node_파이썬_같음(only_missing):
    line = _line()
    results = []
    for fill in (follow.fill_structure_coordinates, follow._fill_structure_coordinates_py):
        items, previous = _items(line)
        count = fill(items, line, previous=previous, only_missing=only_missing)
        results.append((count, [(item.x, item.y) for item in items]))
    assert results[0] == results[1]
    assert results[0][0] > 0


def test_좌표_맞춤_Node_길이_실제로_탐(monkeypatch):
    called = []
    real = follow.run_calc
    monkeypatch.setattr(follow, "run_calc", lambda *a: called.append(a[0]) or real(*a))
    items, previous = _items(_line())
    follow.fill_structure_coordinates(items, _line(), previous=previous)
    assert called == [["facility.follow"]]


def test_측점_계획_Node_파이썬_같음():
    import test_322_3_plan_mirror as plan_mirror

    from B05_Profile.B05_Profile_Engine_Sections_Node import node_plan_stations
    from B05_Profile.B05_Profile_Structures_Schema import StructureInstance
    from common_util.common_util_drainage_pipes import parse_pipe_points
    from common_util.common_util_structure_station import plan_stations

    groups = [
        (
            parse_pipe_points(case["pipes"]),
            [StructureInstance.model_validate(s) for s in case["structures"]],
            list(case.get("plus", ())),
        )
        for case in plan_mirror.CASES.values()
    ]
    node = node_plan_stations(groups)
    assert node is not None
    assert node == [plan_stations(*group) for group in groups]


# ── 화면 — 계산 자리 route 「서버」 노선 = 화면 셈 노선(검증_sub3 · Playwright · 저장 안 함) ──

VERIFY = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"  # 검증_sub3(OWNERS.md)

JS_ROUTE = r"""async () => {
  const urls = performance.getEntriesByType('resource').map((e) => e.name);
  const page = await import(urls.find((n) => /design_pipeline_page\.ts/.test(n)));
  const pid = localStorage.getItem('frd_current_project_id');
  const pipe = page.pagePipeline(pid);
  await page.reconcilePagePipeline(pid);
  await pipe.settled();
  const calls = (window.__aisloCalcServer?.calls ?? []).filter((c) => c.id === 'route.polyline');
  const plain = (v) => JSON.parse(JSON.stringify(v ?? null));
  return { route: plain(pipe.output('route')), stations: plain(pipe.output('stations')),
           calls: calls.map((c) => ({ round: c.roundMs, node: c.nodeMs, same: c.same })),
           runs: (window.__aisloPipeline?.runs ?? []).slice(-1).map((r) => r.outcome) };
}"""


@pytest.fixture(scope="module")
def screens():
    from test_511_b1_calc_node_stages import _UP, BASE, JS_DRAFT, _started

    if not BASE or _UP is None or _UP != _started(BASE.replace("5178", "8005")):
        pytest.skip("이 폴더 서버 없음")
    sync = pytest.importorskip("playwright.sync_api")
    cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
    shots = {}
    with sync.sync_playwright() as play:
        chrome = play.chromium.launch(channel="msedge", headless=True)
        try:
            for mode in ("both", "server"):
                context = chrome.new_context(viewport={"width": 1600, "height": 1000})
                login = {k: cred[k] for k in ("email", "password")}
                r = context.request.post(
                    f"{BASE}/api/auth/login/request", data=login, timeout=60000
                )
                assert r.ok, f"로그인 실패 {r.status}"  # 비밀번호는 안 찍음
                if mode == "server":  # 판은 안 고침 — 이 창이 받는 판만 route = server

                    def board(route):
                        body = route.fetch().json()
                        body["sites"]["route"] = "server"
                        route.fulfill(json=body)

                    context.route("**/api/dashboard/calc-sites", board)
                page = context.new_page()
                page.add_init_script(
                    f"localStorage.setItem('frd_current_project_id', '{VERIFY}');"
                    " performance.setResourceTimingBufferSize(20000);"
                )
                page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
                page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
                page.wait_for_timeout(9000)
                shots[mode] = {"saved": page.evaluate(JS_ROUTE)}
                page.evaluate(JS_DRAFT)  # 교각점 하나 옮김(세션 초안 · 창과 함께 버림)
                page.wait_for_timeout(3000)
                shots[mode]["draft"] = page.evaluate(JS_ROUTE)
                context.close()
        finally:
            chrome.close()
    (ROOT / "tmp" / "511b5_screens.json").write_text(json.dumps(shots), encoding="utf-8")
    return shots


@pytest.mark.parametrize("case", ["saved", "draft"])
def test_화면_서버_노선_화면_셈과_같음(screens, case):
    both, server = screens["both"][case], screens["server"][case]
    assert both["runs"] == server["runs"] == ["done"]
    assert len(both["route"]["line"]) > 10
    assert both["route"] == server["route"] and both["stations"] == server["stations"]
    assert both["calls"] == []
    # 저장 노선(초안 없음) = 서버 안 부름 · 초안 = server 판만 서버 노선
    assert bool(server["calls"]) is (case == "draft")


# ── follow 「서버」 — 창구 맥락 · Node route · follow 두 단계 ─────────────────────────


def test_창구_follow_맥락_브라우저_정본칸_버림(monkeypatch):
    from B06_Section import B06_Section_Router_Calc_Context as ctx

    async def context(_pid):
        return {"plan": {"planned": "서버"}, "saved_pipes": ["서버"], "longitudinal": None}

    monkeypatch.setattr(ctx, "follow_context", context)
    inputs = {
        "stages": {
            "until": "follow",
            "input": {"route": 1},
            "saved_edits": None,
            "plan": {"planned": "브라우저"},
            "saved_pipes": ["브라우저"],
            "grid": {"path": "C:/x"},
            "detail": {"x": 1},
        }
    }
    assert asyncio.run(ctx.attach_follow_context("p", inputs)) is None
    assert inputs["stages"] == {
        "until": "follow",
        "input": {"route": 1},
        "saved_edits": None,
        "plan": {"planned": "서버"},
        "saved_pipes": ["서버"],
        "longitudinal": None,
    }
    other = {"stages": {"plan": "브라우저"}}  # until 없음 = 그대로(B-1 단계 길)
    assert asyncio.run(ctx.attach_follow_context("p", other)) is None and other["stages"] == {
        "plan": "브라우저"
    }


def _golden_stages() -> dict:
    from common_util.common_util_surface_sampler import dtm_grid_bytes

    def fixture(name):
        return json.loads((FIXTURE / name).read_text(encoding="utf-8"))

    plan = fixture("plan.json")
    nodes = [[n["x"], n["y"]] for n in plan["nodes"]]
    at = len(nodes) // 2
    nodes[at] = [nodes[at][0] + 6, nodes[at][1] + 4]  # 교각점 하나 옮긴 노선 초안
    draft = {
        "nodes": nodes,
        "curveOn": [n["radius_m"] is not None for n in plan["nodes"]],
        "curveRadius": [None] * len(nodes),
        "curveLock": [None] * len(nodes),
        "curveArc": [None] * len(nodes),
        "stationSnap": bool(plan.get("station_snap")),
        "stationSnapStep": plan.get("station_snap_step"),
    }
    grid = ROOT / "tmp" / "511b5_grid.bin"
    if not grid.is_file():
        grid.write_bytes(
            dtm_grid_bytes(FIXTURE / "project/B04_PreProcess/models/dtm_classification_smooth.npz")
        )
    return {
        "plan": plan,
        "detail": fixture("detail.json"),
        "types": fixture("structure_types.json")["types"],
        "saved_pipes": [
            {k: v for k, v in p.items() if k != "lonlat"}
            for p in fixture("pipes.json")["pipe_points"]
        ],
        "saved_structures": fixture("structures.json")["structures"],
        "interval_m": 20,
        "grid": {"path": str(grid), "design": "las"},
        "input": {"route": draft, "plusStations": fixture("plus_stations.json")["chainages"]},
    }


def test_Node_follow_두_단계_묶음_첫_바퀴와_같음():
    from common_util import common_util_calc_runner as runner

    stages = _golden_stages()
    full = runner.run_calc_served(["stage.route", "stage.follow"], {"stages": stages})
    only = runner.run_calc_served(
        ["stage.route", "stage.follow"], {"stages": {**stages, "until": "follow", "grid": None}}
    )
    assert full and only and "error" not in full and "error" not in only, (full, only)
    assert only["results"]["stage.route"] == full["results"]["stage.route"]
    got = only["results"]["stage.follow"]
    assert got["patch"] == full["results"]["stage.follow"]
    assert got["patch"]["pipes"] and got["followed"]  # 관 따라감 · 따라가기만 한 자리 곁 자료


# ── 화면 — follow 「서버」 끌기 여러 걸음 뒤 놓음 = 화면 셈(검증_sub3 · Playwright · 저장 안 함) ──

JS_DRAG = r"""async (steps) => {
  const urls = performance.getEntriesByType('resource').map((e) => e.name);
  const mod = async (re) => import(urls.find((n) => re.test(n)));
  const page = await mod(/design_pipeline_page\.ts/);
  const pid = localStorage.getItem('frd_current_project_id');
  const plan = await (await mod(/B05_Profile_Api_Replan\.ts/)).fetchRoutePlan(pid);
  const flat = (await mod(/B05_Profile_UI_RouteEdit_Edits\.ts/))
    .flattenServerPlan(plan.nodes, plan.curves ?? [], plan.node_edits);
  const bridge = page.routeDraftBridge(pid, null);
  const at = Math.floor(flat.planned.length / 2);
  for (let k = 1; k <= steps; k += 1) {  // 끌기 걸음 — 걸음마다 묶음이 돎(잠잠 안 기다림)
    const nodes = flat.planned.map(([x, y]) => [x, y]);
    nodes[at] = [nodes[at][0] + 2 * k, nodes[at][1] + 1.5 * k];
    bridge.edited({
      nodes,
      curveOn: flat.curveOn.map((on) => on !== false),
      curveRadius: flat.curveRadius.map((r) => r ?? null),
      curveLock: flat.curveLock ?? nodes.map(() => null),
      curveArc: flat.curveArc ?? nodes.map(() => null),
      stationSnap: plan.station_snap ?? false,
      stationSnapStep: plan.station_snap_step ?? null,
    });
    await new Promise((r) => setTimeout(r, 60));
  }
  await page.pagePipeline(pid).settled();
}"""

JS_FOLLOW = r"""async () => {
  const urls = performance.getEntriesByType('resource').map((e) => e.name);
  const page = await import(urls.find((n) => /design_pipeline_page\.ts/.test(n)));
  const pid = localStorage.getItem('frd_current_project_id');
  const pipe = page.pagePipeline(pid);
  await page.reconcilePagePipeline(pid);
  await pipe.settled();
  const plain = (v) => JSON.parse(JSON.stringify(v ?? null));
  const calls = (window.__aisloCalcServer?.calls ?? []).filter((c) => c.id === 'stage.follow');
  const planline = pipe.output('planline');
  return {
    follow: plain(pipe.output('follow')),
    inputs: plain(['pipes', 'structures', 'plusStations', 'alignment']
      .map((k) => [k, pipe.input(k)])),
    base_pvi: plain(planline?.basePvi), pvi: plain(pipe.output('profile')?.pvi),
    calls: calls.map((c) => ({ round: c.roundMs, node: c.nodeMs, bytes: c.bytes })),
    runs: (window.__aisloPipeline?.runs ?? []).slice(-1).map((r) => r.outcome),
  };
}"""


@pytest.fixture(scope="module")
def drags():
    from test_511_b1_calc_node_stages import _UP, BASE, _started

    if not BASE or _UP is None or _UP != _started(BASE.replace("5178", "8005")):
        pytest.skip("이 폴더 서버 없음")
    sync = pytest.importorskip("playwright.sync_api")
    cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
    shots = {}
    with sync.sync_playwright() as play:
        chrome = play.chromium.launch(channel="msedge", headless=True)
        try:
            for mode in ("both", "server"):
                context = chrome.new_context(viewport={"width": 1600, "height": 1000})
                login = {k: cred[k] for k in ("email", "password")}
                r = context.request.post(
                    f"{BASE}/api/auth/login/request", data=login, timeout=60000
                )
                assert r.ok, f"로그인 실패 {r.status}"  # 비밀번호는 안 찍음
                if mode == "server":  # 판은 안 고침 — 이 창이 받는 판만 follow = server

                    def board(route):
                        body = route.fetch().json()
                        body["sites"]["follow"] = "server"
                        route.fulfill(json=body)

                    context.route("**/api/dashboard/calc-sites", board)
                page = context.new_page()
                page.add_init_script(
                    f"localStorage.setItem('frd_current_project_id', '{VERIFY}');"
                    " performance.setResourceTimingBufferSize(20000);"
                )
                page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
                page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
                page.wait_for_timeout(9000)
                page.evaluate(JS_DRAG, 4)  # 끌기 네 걸음(세션 초안 · 창과 함께 버림)
                page.wait_for_timeout(4000)  # 잠잠 400 ms · 서버 · 다시 돎
                shots[mode] = page.evaluate(JS_FOLLOW)
                context.close()
        finally:
            chrome.close()
    (ROOT / "tmp" / "511b5_drags.json").write_text(json.dumps(shots), encoding="utf-8")
    return shots


def test_화면_follow_서버_끌기_뒤_화면_셈과_같음(drags):
    both, server = drags["both"], drags["server"]
    assert both["runs"] == server["runs"] == ["done"]
    assert both["calls"] == [] and server["calls"]  # server 판만 놓은 뒤 서버 follow
    # 계획고 옮김 = 끌기 중 화면 셈이 입력에 옮겨 둔 값 그대로(서버 follow 는 저장분에서 새로 셈 · 옮김
    # 기준(앞 노선 · 앞 첫 계획선) 없음 → patch 에 alignment 안 냄) — 입력 · 종단은 같아야 함
    drop = lambda patch: {k: v for k, v in patch.items() if k != "alignment"}  # noqa: E731
    assert drop(server["follow"]) == drop(both["follow"])
    for key in ("inputs", "base_pvi", "pvi"):
        assert server[key] == both[key], key