Aislo/B08_Quantity/B08_Quantity_Milestone_Capture.py

"""B08 거리이정표 경로 그림 자동 캡처(PLAN 244-11 · 사용자 139번) — 서버 헤드리스 Edge.

[등록] · [길찾기] 뒤 화면이 그 경로의 길 선(마지막 길찾기 경로 좌표) · 표지를 보냄 →
서버가 헤드리스 Edge(Playwright channel msedge · 없으면 chrome · 번들 chromium 안 씀)로
카카오 지도 JS + 길 선 + 표지 + 경로 맞춤 → tilesloaded 뒤 1200×800 PNG → `<폴더>/<경로 id>.png`
(사용자 그림이 있으면 화면은 사용자 그림 우선).
페이지 주소 = .env KAKAO_CAPTURE_ORIGIN(없으면 http://localhost:5173) + /__route_capture —
page.route 로 가로채 채움(카카오 JS 키 도메인 검사만 통과 · vite 안 떠도 됨).
경로 좌표는 그림 만든 뒤 버림(저장 = 그림 · km) · 동시 요청 = 하나씩(잠금) · 실패 = 까닭 한 줄 ·
키 값은 돌려주지도 적지도 않음.
"""

from __future__ import annotations

import json
import os
import threading
import time
import urllib.parse
from pathlib import Path
from typing import Any

KAKAO_JS = "KAKAO_JS_KEY"
ORIGIN_KEY = "KAKAO_CAPTURE_ORIGIN"
DEFAULT_ORIGIN = "http://localhost:5173"
PAGE = "/__route_capture"
WIDTH, HEIGHT = 1200, 800
TIMEOUT_S = 20
CHANNELS = ("msedge", "chrome")  # 시스템 브라우저 — 번들 chromium 은 안 받아 둠
LINE_COLOR = "#E60046"  # 화면 지도(`…_UI_Milestone_KakaoMap`) 길 선과 같은 색
TAIL = " — 그림 올리기를 쓰세요."

_lock = threading.Lock()  # ponytail: 서버 전체 한 줄 — 브라우저 한 번에 하나(2~3초)


class CaptureError(Exception):
    """까닭 한 줄(키 · 예외 원문 없음)."""


def page_html(key: str, lines: list[Any], marks: list[dict[str, Any]]) -> str:
    """카카오 지도 + 길 선 + 표지 + 경로 맞춤 · 다 그림 = window.__done · SDK 못 받음 = __fail."""
    data = json.dumps({"lines": lines, "marks": marks}, ensure_ascii=False).replace("<", "\\u003c")
    sdk = f"https://dapi.kakao.com/v2/maps/sdk.js?appkey={urllib.parse.quote(key)}&autoload=false"
    return f"""<!doctype html><html><head><meta charset="utf-8"></head><body style="margin:0">
<div id="m" style="width:{WIDTH}px;height:{HEIGHT}px"></div>
<script src="{sdk}" onerror="window.__fail=1"></script>
<script>window.kakao && kakao.maps.load(() => {{
  const D = {data}, K = kakao.maps, at = (p) => new K.LatLng(p[0], p[1]);
  const pts = [...D.marks.map((m) => [m.lat, m.lon]), ...D.lines.flat()].map(at);
  const map = new K.Map(document.getElementById("m"), {{ center: pts[0], level: 7 }});
  D.lines.forEach((l) => new K.Polyline({{ map, path: l.map(at), strokeWeight: 6,
    strokeColor: "{LINE_COLOR}", strokeOpacity: 0.9 }}));
  D.marks.forEach((m) => {{
    const p = at([m.lat, m.lon]);
    new K.Marker({{ map, position: p, title: m.name }});
    const s = document.createElement("span");
    s.textContent = m.name;
    s.style.cssText = "background:#fff;border:1px solid #333;border-radius:4px;padding:2px 6px;"
      + "font:600 13px sans-serif;white-space:nowrap";
    new K.CustomOverlay({{ map, position: p, content: s, yAnchor: 2.6 }});
  }});
  K.event.addListener(map, "tilesloaded", () => (window.__done = true));
  const b = new K.LatLngBounds();
  pts.forEach((p) => b.extend(p));
  map.setBounds(b, 60, 60, 60, 60);
}});</script></body></html>"""


def shoot(html: str, origin: str) -> bytes:
    """헤드리스 브라우저로 그 페이지 → 지도 칸 PNG. 실패 = CaptureError(까닭 한 줄)."""
    try:
        from playwright.sync_api import Error as PwError
        from playwright.sync_api import TimeoutError as PwTimeout
        from playwright.sync_api import sync_playwright
    except ImportError as error:
        raise CaptureError("서버에 Playwright 가 없습니다(pip install playwright)") from error
    url = origin + PAGE
    deadline = time.monotonic() + TIMEOUT_S

    def left() -> float:
        return max(1.0, deadline - time.monotonic()) * 1000

    try:
        with sync_playwright() as p:
            browser = None
            for channel in CHANNELS:
                try:
                    browser = p.chromium.launch(channel=channel, headless=True, timeout=left())
                    break
                except PwError:
                    continue
            if browser is None:
                raise CaptureError("서버에서 브라우저(Edge · Chrome)를 못 띄움")
            try:
                page = browser.new_page(viewport={"width": WIDTH, "height": HEIGHT})
                page.route(
                    url,
                    lambda route: route.fulfill(
                        status=200, content_type="text/html; charset=utf-8", body=html
                    ),
                )
                page.goto(url, timeout=left())
                page.wait_for_function("window.__done || window.__fail", timeout=left())
                if page.evaluate("window.__fail"):
                    raise CaptureError(
                        f"카카오 지도 JS 가 캡처 주소({origin})를 받지 않음(키 도메인 등록 확인)"
                    )
                page.wait_for_timeout(500)  # 마지막 타일 · 표지 그림
                return page.locator("#m").screenshot(timeout=left())
            finally:
                browser.close()
    except CaptureError:
        raise
    except PwTimeout as error:
        raise CaptureError(f"지도 그림이 {TIMEOUT_S}초 안에 안 끝남") from error
    except Exception as error:  # 드라이버 못 뜸 등 — 예외 글은 내지 않음
        raise CaptureError("서버에서 브라우저(Edge · Chrome)를 못 띄움") from error


def make_route_image(
    out_dir: Path, route_id: str, lines: list[Any], marks: list[dict[str, Any]]
) -> dict[str, Any]:
    """그 경로 자동 그림 → {saved, reason} — 좌표는 그림에만 쓰고 버림."""
    key = os.environ.get(KAKAO_JS, "").strip()
    if not key:
        return {
            "saved": False,
            "reason": f"카카오 지도 JavaScript 키가 없습니다(.env {KAKAO_JS}){TAIL}",
        }
    if not any(lines) and not marks:
        return {"saved": False, "reason": f"길 선 · 표지가 없어 지도 그림을 못 만듦{TAIL}"}
    origin = (os.environ.get(ORIGIN_KEY, "").strip() or DEFAULT_ORIGIN).rstrip("/")
    with _lock:
        try:
            png = shoot(page_html(key, lines, marks), origin)
        except CaptureError as error:
            return {"saved": False, "reason": f"경로 그림 자동 캡처 실패 — {error}{TAIL}"}
        out_dir.mkdir(parents=True, exist_ok=True)
        new = out_dir / f".{route_id}.png.new"
        new.write_bytes(png)
        new.replace(out_dir / f"{route_id}.png")
    return {"saved": True, "reason": None}