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) · 동시 요청 = 하나씩(잠금) · 실패 = 까닭 한 줄 ·
키 값은 돌려주지도 적지도 않음.
캡처는 자식 파이썬 프로세스(이 파일을 스크립트로 · 입력 = JSON stdin · 출력 = 결과 JSON 한 줄 ·
키는 환경 그대로) — 서버(uvicorn · Windows Selector 루프) 안 스레드에선 Playwright 가
드라이버 하위 프로세스를 못 만듦(249-1).
장소 그림(256-1) = 같은 길 · 작은 지도(480×320 · 표지 하나 · 주소 둘레 줌) → `place/<상자 id>.png`.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
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
CHILD_TIMEOUT_S = 25  # 자식 프로세스 전체(파이썬 · 드라이버 기동 포함)
CHILD = __file__  # 자식으로 돌릴 스크립트
NO_BROWSER = "서버에서 브라우저(Edge · Chrome)를 못 띄움"
CHANNELS = ("msedge", "chrome")  # 시스템 브라우저 — 번들 chromium 은 안 받아 둠
LINE_COLOR = "#E60046"  # 화면 지도(`…_UI_Milestone_KakaoMap`) 길 선과 같은 색
TAIL = " — [그림 변경]으로 넣으세요."
PLACE_VIEW = {"w": 480, "h": 320, "level": 4}  # 장소 그림(256-1) — 표지 가운데 · 고정 줌

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


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


def page_html(
    key: str, lines: list[Any], marks: list[dict[str, Any]], view: dict[str, int] | None = None
) -> str:
    """카카오 지도 + 길 선 + 표지 + 경로 맞춤 · 다 그림 = window.__done · SDK 못 받음 = __fail.
    view = 크기 · 줌 고정(장소 그림 · 첫 표지 가운데 · 경로 맞춤 안 함)."""
    w, h = (view["w"], view["h"]) if view else (WIDTH, HEIGHT)
    level = view["level"] if view else 7
    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:{w}px;height:{h}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: {level} }});
  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));
  if ({"false" if view else "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(NO_BROWSER)
            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(NO_BROWSER) from error


def capture(job: dict[str, Any]) -> dict[str, Any]:
    """자식 프로세스 몸통 — 그림 찍어 `<out>/<route_id>.png` 저장 → {saved, reason}."""
    key = os.environ.get(KAKAO_JS, "").strip()
    try:
        png = shoot(page_html(key, job["lines"], job["marks"], job.get("view")), job["origin"])
    except CaptureError as error:
        return {"saved": False, "reason": str(error)}
    out_dir = Path(job["out"])
    out_dir.mkdir(parents=True, exist_ok=True)
    new = out_dir / f".{job['route_id']}.png.new"
    new.write_bytes(png)
    new.replace(out_dir / f"{job['route_id']}.png")
    return {"saved": True, "reason": None}


def spawn(job: dict[str, Any]) -> dict[str, Any]:
    """자식 파이썬 프로세스로 capture — 서버 이벤트 루프 정책과 무관(기본 Proactor)."""
    try:
        done = subprocess.run(
            [sys.executable, CHILD],
            input=json.dumps(job).encode(),
            capture_output=True,
            timeout=CHILD_TIMEOUT_S,
        )
    except subprocess.TimeoutExpired:
        return {"saved": False, "reason": f"지도 그림이 {TIMEOUT_S}초 안에 안 끝남"}
    except OSError:
        return {"saved": False, "reason": NO_BROWSER}
    try:
        return json.loads(done.stdout.decode().strip().splitlines()[-1])
    except (ValueError, IndexError):  # 자식이 죽음 — stderr(예외 글)는 내지 않음
        return {"saved": False, "reason": NO_BROWSER}


def make_route_image(
    out_dir: Path,
    route_id: str,
    lines: list[Any],
    marks: list[dict[str, Any]],
    view: dict[str, int] | None = None,
    what: str = "경로 그림",
    tail: str = TAIL,
) -> 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("/")
    job = {"out": str(out_dir), "route_id": route_id, "lines": lines, "marks": marks, "view": view}
    with _lock:
        got = spawn({**job, "origin": origin})
    if not got.get("saved"):
        return {"saved": False, "reason": f"{what} 자동 캡처 실패 — {got.get('reason')}{tail}"}
    return {"saved": True, "reason": None}


def make_place_image(out_dir: Path, place_id: str, mark: dict[str, Any]) -> dict[str, Any]:
    """장소 그림(256-1) — 작은 지도 · 표지 하나 → `<out_dir>/<상자 id>.png`."""
    return make_route_image(out_dir, place_id, [], [mark], PLACE_VIEW, "장소 그림", "")


if __name__ == "__main__":
    print(json.dumps(capture(json.loads(sys.stdin.buffer.read()))))