# -*- coding: utf-8 -*-
"""508-3b · c 화면 — 자기 포트(5177) Playwright 머리 없는 브라우저 · 검증_sub2 · `AISLO_LIVE_DB=1` 일 때만.

- B07 열기 → 미리받기 차례 = 가벼운 것 먼저 · 평면도(무거운 것) 끝
- B06 으로 가 첫 측점 깎기 비탈 고치고 [저장] → B07 돌아옴 → 지문 바뀐 도면만 다시 받음
  (횡단 = 고친 측점 든 장 하나 · 측점 지문도 그 측점만) · 페이지 옮겨도 받은 것은 그대로
- 안 받은 평면도(유역도 · 저장본 지움) 열기 → 「받는 중」 → 받은 뒤 보임
- 평면도 둘째 열기 = 서버 저장본(시간 전 · 후) · GET 이 지우는 파일 없음
- 못 받은 도면 = 「다시 받기」 → 누르면 다시 받음(밝음 · 어두움 한 장씩 · tmp/shots_508_3)
- 단계 행 · projects · 노선 · 종 · 횡단 행 · B05 노선 · B06 · B07 storage 를 먼저 떠 두고 끝에 되돌림

  AISLO_LIVE_DB=1 ./venv/Scripts/python.exe -m pytest resources/tester/test_508_3_b07_prefetch_browser.py -q -s
"""

from __future__ import annotations

import asyncio
import json
import os
import shutil
import sys
import time
import urllib.request
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
from helper_verify_project import own_ports  # noqa: E402

BASE = own_ports()[1]  # 화면 서버 = OWNERS.md 이 폴더 줄
PROJECT = "cb15892a-5788-4a26-9316-20a15421f68b"  # 검증_sub2(OWNERS.md)
SHOTS = ROOT / "tmp" / "shots_508_3"
STAGE_COLS = (
    "state",
    "progress_percent",
    "params",
    "message",
    "started_at",
    "completed_at",
    "updated_at",
)
KEEP_DIRS = ("B05_Profile/route", "B06_Section", "B07_DesignDetail")


def _up() -> bool:
    try:
        urllib.request.urlopen(BASE, timeout=3)
        return True
    except Exception:
        return False


pytestmark = pytest.mark.skipif(
    os.environ.get("AISLO_LIVE_DB") != "1" or not _up(),
    reason="AISLO_LIVE_DB=1 · 5177 서버 있을 때만(검증_sub2 storage · DB 행을 씀)",
)


async def _keep(mode: str, held: dict) -> None:
    """단계 · projects · routes · 종 · 횡단 행 + storage 폴더 떠 두기(save) · 되돌리기(restore)."""
    import aiomysql

    from common_util.common_util_storage import resolve_stored_project_path
    from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER

    conn = await aiomysql.connect(
        host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, db=DB_NAME,
        charset="utf8mb4", autocommit=False,
    )  # fmt: skip
    try:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            if mode == "save":
                await cur.execute("SELECT storage_path FROM projects WHERE id=%s", (PROJECT,))
                held["root"] = Path(
                    resolve_stored_project_path((await cur.fetchone())["storage_path"])
                )
                for name, sql in (
                    ("stages", "SELECT * FROM project_workflow_stages WHERE project_id=%s"),
                    ("project", "SELECT status, updated_at FROM projects WHERE id=%s"),
                    ("routes", "SELECT id, status FROM routes WHERE project_id=%s"),
                    ("long", "SELECT id, data, status FROM longitudinal_sections WHERE project_id=%s"),
                    ("cross", "SELECT * FROM cross_sections WHERE project_id=%s"),
                ):  # fmt: skip
                    await cur.execute(sql, (PROJECT,))
                    held[name] = await cur.fetchall()
                held["copy"] = ROOT / "tmp" / f"pt_508_3_keep_{int(time.time())}"
                for rel in KEEP_DIRS:
                    shutil.copytree(held["root"] / rel, held["copy"] / rel)
                return
            for row in held["stages"]:
                await cur.execute(
                    f"UPDATE project_workflow_stages SET {', '.join(f'{k}=%s' for k in STAGE_COLS)} "
                    "WHERE id=%s",
                    (*[row[k] for k in STAGE_COLS], row["id"]),
                )
            project = held["project"][0]
            await cur.execute(
                "UPDATE projects SET status=%s, updated_at=%s WHERE id=%s",
                (project["status"], project["updated_at"], PROJECT),
            )
            for row in held["routes"]:
                await cur.execute(
                    "UPDATE routes SET status=%s WHERE id=%s", (row["status"], row["id"])
                )
            for row in held["long"]:
                await cur.execute(
                    "UPDATE longitudinal_sections SET data=%s, status=%s WHERE id=%s",
                    (row["data"], row["status"], row["id"]),
                )
            await cur.execute("DELETE FROM cross_sections WHERE project_id=%s", (PROJECT,))
            for row in held["cross"]:
                cols = list(row)
                await cur.execute(
                    f"INSERT INTO cross_sections ({', '.join(cols)}) "
                    f"VALUES ({', '.join(['%s'] * len(cols))})",
                    [row[c] for c in cols],
                )
        await conn.commit()
        for rel in KEEP_DIRS:
            shutil.rmtree(held["root"] / rel)
            shutil.copytree(held["copy"] / rel, held["root"] / rel)
    finally:
        conn.close()


@pytest.fixture(scope="module")
def ctx():
    sync = pytest.importorskip("playwright.sync_api")
    cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
    held: dict = {}
    asyncio.run(_keep("save", held))
    try:
        with sync.sync_playwright() as play:
            chrome = play.chromium.launch(channel="msedge", headless=True)
            context = chrome.new_context(viewport={"width": 1600, "height": 950})
            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}"  # 비밀번호는 안 찍음
            context.add_init_script(f"localStorage.setItem('frd_current_project_id', '{PROJECT}')")
            context.held = held  # type: ignore[attr-defined]
            yield context
            chrome.close()
    finally:
        asyncio.run(_keep("restore", held))
        print("되돌림 사본 자리", held.get("copy"))


def _list(ctx) -> dict:
    body = ctx.request.get(f"{BASE}/api/projects/{PROJECT}/design-drawings", timeout=120000)
    return {item["id"]: item for item in body.json()["drawings"]}


def _watch(page) -> list[tuple[float, str]]:
    gets: list[tuple[float, str]] = []
    page.on(
        "requestfinished",
        lambda q: (
            q.method == "GET"
            and "/design-drawings/" in q.url
            and gets.append((time.time(), q.url.rsplit("/", 1)[-1]))
        ),
    )
    return gets


def _settle(page, gets: list, want: set[str], start: int = 0, timeout_s: float = 240) -> None:
    """`start` 뒤로 원하는 도면을 다 받고 5 s 동안 새 GET 이 없을 때까지."""
    end = time.time() + timeout_s
    while time.time() < end:
        got = {name for _t, name in gets[start:]}
        quiet = len(gets) == start or time.time() - gets[-1][0] > 5
        if want <= got and quiet:
            return
        page.wait_for_timeout(500)
    raise AssertionError(f"못 받음 {sorted(want - {name for _t, name in gets[start:]})}")


def _b07(page) -> None:
    page.wait_for_selector(".b07-drawing-button[data-active='true']", timeout=90000)


def _open_group(page, kind: str) -> None:
    """접힌 목록 묶음을 폄."""
    group = page.locator(f".b07-drawing-group[data-kind='{kind}']")
    if "is-collapsed" in (group.get_attribute("class") or ""):
        group.locator(".ui-collapsible__title").first.click()


def _cut_slope_and_save(page, value: str) -> str:
    """B06 첫 횡단 카드 깎기 비탈 → [저장] · 카드 머리 글을 돌려줌."""
    box = page.locator(".b06-design__cutslope-input").first
    box.wait_for(timeout=120000)
    page.wait_for_timeout(3000)
    title = page.locator("article.b06-cross-card").first.inner_text().split("\n")[0]
    box.fill(value)
    box.press("Enter")
    page.wait_for_timeout(800)
    with page.expect_response(
        lambda r: "/design/save" in r.url and r.request.method == "POST", timeout=180000
    ) as saved:
        page.locator("button.ui-btn", has_text="저장").last.click()
    assert saved.value.ok, f"B06 저장 실패 {saved.value.status}"
    page.wait_for_timeout(2000)
    return title


def test_b07_미리받기_지문_화면(ctx):
    root: Path = ctx.held["root"]
    SHOTS.mkdir(parents=True, exist_ok=True)
    page = ctx.new_page()
    gets = _watch(page)
    # 옛 판 자료는 첫 B06 [저장]이 모든 측점 설계에 새 칸을 붙임(한 번만 · 실제 자료 바뀜) — 먼저 한 번
    page.goto(f"{BASE}/#/b06-section", wait_until="domcontentloaded")
    _cut_slope_and_save(page, "0.45")
    before = _list(ctx)
    mark = len(gets)
    page.evaluate("location.hash = '#/b07-design-detail'")
    _b07(page)
    _settle(page, gets, set(before), mark)
    order = [name for _t, name in gets[mark:]]
    heavy = [i for i, name in enumerate(order) if before[name]["kind"] == "plan"]
    light = [i for i, name in enumerate(order) if before[name]["kind"] != "plan"]
    print("미리받기 차례", order)
    assert len(order) == len(set(order)) == len(before)  # 한 장 한 번
    assert max(light) < min(heavy)  # 무거운 것 끝

    # B06 으로 가 측점 하나 고치고 [저장] → B07 로 돌아옴(같은 창 · 페이지 밖 모듈 그대로)
    page.evaluate("location.hash = '#/b06-section'")
    card = _cut_slope_and_save(page, "0.55")
    after = _list(ctx)
    changed = {key for key in before if before[key]["fingerprint"] != after[key]["fingerprint"]}
    sheets = {key for key in changed if before[key]["kind"] == "cross"}
    stations = {
        (key, c)
        for key in sheets
        for c, value in (after[key]["stations"] or {}).items()
        if before[key]["stations"].get(c) != value
    }
    print(f"고친 카드 「{card}」 · 지문 바뀐 도면 {sorted(changed)} · 측점 {sorted(stations)}")
    assert len(sheets) == 1 and len(stations) == 1  # 그 장만 · 그 측점만
    mark = len(gets)
    page.evaluate("location.hash = '#/b07-design-detail'")
    _b07(page)
    _settle(page, gets, changed, mark)
    again = [name for _t, name in gets[mark:]]
    print("돌아와 다시 받음", again)
    assert set(again) == changed and len(again) == len(changed)

    # 되돌림(화면 조작으로 바꾼 값 = 원래 값) — 행 · 파일은 끝에 통째로 되돌림
    page.evaluate("location.hash = '#/b06-section'")
    _cut_slope_and_save(page, "0.40")
    page.close()

    # 평면도 둘째 열기 = 서버 저장본(시간 전 · 후) · GET 이 지우는 파일 없음
    shutil.rmtree(root / "_derived" / "b07_plan_plan_watershed", ignore_errors=True)
    shutil.rmtree(root / "_derived" / "b07_plan_plan_terrain", ignore_errors=True)
    files = {p for p in root.rglob("*") if p.is_file() and "B05_Profile" not in p.parts}
    url = f"{BASE}/api/projects/{PROJECT}/design-drawings/plan_terrain"
    spent = []
    for _ in range(2):
        start = time.time()
        assert ctx.request.get(url, timeout=120000).ok
        spent.append(time.time() - start)
    print(f"평면도(지형) 첫 열기 {spent[0]:.2f} s → 둘째(저장본) {spent[1]:.2f} s")
    assert spent[1] < spent[0]
    gone = {p for p in files if not p.exists()}
    assert not gone, f"GET 이 지움 {sorted(map(str, gone))[:5]}"
    assert list((root / "_derived" / "b07_plan_plan_terrain").glob("*.json"))

    # 안 받은 평면도(유역도 · 저장본 없음) 열기 → 「받는 중」 → 받은 뒤 보임
    page = ctx.new_page()
    page.goto(f"{BASE}/#/b07-design-detail", wait_until="domcontentloaded")
    _b07(page)
    _open_group(page, "plan")
    start = time.time()
    page.locator(".b07-drawing-button[data-drawing-id='plan_watershed']").click()
    page.wait_for_selector(".ui-toast:has-text('받는 중')", timeout=10000)
    page.wait_for_function(
        "() => document.querySelector('.b07-drawing-button[data-drawing-id=plan_watershed]')"
        "?.dataset.fetch === 'ready' && document.querySelector('.cad-host')?.dataset.loading === 'false'",
        timeout=120000,
    )
    print(f"유역도 받는 중 → 보임 {time.time() - start:.2f} s")
    page.screenshot(path=str(SHOTS / "watershed_light.png"))
    page.evaluate("document.documentElement.setAttribute('data-theme', 'dark')")
    page.wait_for_timeout(500)
    page.screenshot(path=str(SHOTS / "watershed_dark.png"))
    page.evaluate("document.documentElement.setAttribute('data-theme', 'light')")
    page.close()


def test_b07_다시_받기(ctx):
    page = ctx.new_page()
    target = "mass_haul"
    fails = {"n": 0}

    def fail(route):
        fails["n"] += 1
        route.abort()

    page.route(f"**/design-drawings/{target}", fail)
    page.goto(f"{BASE}/#/b07-design-detail", wait_until="domcontentloaded")
    _b07(page)
    button = page.locator(f".b07-drawing-button[data-drawing-id='{target}']")
    page.wait_for_function(
        f"() => document.querySelector('.b07-drawing-button[data-drawing-id={target}]')"
        "?.dataset.fetch === 'error'",
        timeout=120000,
    )
    print("미리받기 실패 횟수(한 번 다시 포함)", fails["n"])
    assert fails["n"] == 2
    text = page.evaluate(
        f"getComputedStyle(document.querySelector('.b07-drawing-button[data-drawing-id={target}]"
        " .b07-drawing-button__name'), '::after').content"
    )
    assert "다시 받기" in text
    panel = button.locator("xpath=..")
    panel.screenshot(path=str(SHOTS / "retry_light.png"))
    page.evaluate("document.documentElement.setAttribute('data-theme', 'dark')")
    panel.screenshot(path=str(SHOTS / "retry_dark.png"))
    page.evaluate("document.documentElement.setAttribute('data-theme', 'light')")
    page.unroute(f"**/design-drawings/{target}")
    button.click()
    page.wait_for_function(
        f"() => document.querySelector('.b07-drawing-button[data-drawing-id={target}]')"
        "?.dataset.fetch === 'ready' && document.querySelector('.cad-host')?.dataset.loading === 'false'",
        timeout=60000,
    )
    page.screenshot(path=str(SHOTS / "retry_after.png"))
    page.close()