web-page-backend/insta-lab/app/card_renderer.py

"""Jinja → HTML → Playwright headless screenshot."""

import asyncio
import hashlib
import json
import logging
import os
import tempfile
from pathlib import Path
from typing import List

from jinja2 import Environment, FileSystemLoader, select_autoescape
from playwright.async_api import async_playwright

from .config import CARDS_DIR, CARD_TEMPLATE_DIR
from . import db

logger = logging.getLogger(__name__)

# NAS Celeron 2C 환경에서 Chromium을 동시에 여러 인스턴스로 띄우면 CPU/메모리 폭주.
# 슬레이트 렌더는 디스크 I/O와 Chromium launch가 직렬화되어도 충분히 빠르므로
# 단일 슬롯으로 직렬화한다. (CHECK_POINT FU-C)
_RENDER_SEMAPHORE: asyncio.Semaphore | None = None


def _render_semaphore() -> asyncio.Semaphore:
    global _RENDER_SEMAPHORE
    if _RENDER_SEMAPHORE is None:
        _RENDER_SEMAPHORE = asyncio.Semaphore(1)
    return _RENDER_SEMAPHORE


# Chromium 브라우저 풀 — 매 슬레이트마다 launch 하지 않고 1개를 살려둠.
# (CHECK_POINT 중기-6) 카드 10장 렌더 시간 ~30% 단축 기대.
_PLAYWRIGHT = None
_BROWSER = None


async def init_browser() -> None:
    """앱 startup hook에서 1회 호출. 이미 살아있으면 no-op."""
    global _PLAYWRIGHT, _BROWSER
    if _BROWSER is not None and _BROWSER.is_connected():
        return
    _PLAYWRIGHT = await async_playwright().start()
    _BROWSER = await _PLAYWRIGHT.chromium.launch()
    logger.info("Chromium browser pool 초기화 완료")


async def shutdown_browser() -> None:
    """앱 shutdown hook에서 1회 호출."""
    global _PLAYWRIGHT, _BROWSER
    if _BROWSER is not None:
        try:
            await _BROWSER.close()
        except Exception:
            logger.exception("browser close 중 예외 (무시)")
        _BROWSER = None
    if _PLAYWRIGHT is not None:
        try:
            await _PLAYWRIGHT.stop()
        except Exception:
            logger.exception("playwright stop 중 예외 (무시)")
        _PLAYWRIGHT = None


async def _get_browser():
    """현재 브라우저 핸들 반환. crashed/None이면 재초기화 후 반환."""
    global _BROWSER
    if _BROWSER is None or not _BROWSER.is_connected():
        await init_browser()
    return _BROWSER


def _resolve_template_dir() -> str:
    """Prefer config CARD_TEMPLATE_DIR if it exists; else fall back to in-repo templates/."""
    if os.path.isdir(CARD_TEMPLATE_DIR):
        return CARD_TEMPLATE_DIR
    return os.path.join(os.path.dirname(__file__), "templates")


def _env() -> Environment:
    return Environment(
        loader=FileSystemLoader(_resolve_template_dir()),
        autoescape=select_autoescape(["html", "j2"]),
    )


def _slate_dir(slate_id: int) -> str:
    out = os.path.join(CARDS_DIR, str(slate_id))
    os.makedirs(out, exist_ok=True)
    return out


def _build_pages(slate: dict) -> List[dict]:
    cover = json.loads(slate["cover_copy"] or "{}")
    bodies = json.loads(slate["body_copies"] or "[]")
    cta = json.loads(slate["cta_copy"] or "{}")
    accent = cover.get("accent_color") or "#0F62FE"
    pages: List[dict] = []
    pages.append({
        "page_type": "cover", "page_no": 1, "total_pages": 10,
        "headline": cover.get("headline", ""), "body": cover.get("body", ""),
        "accent_color": accent, "cta": "",
    })
    for i, b in enumerate(bodies[:8]):
        pages.append({
            "page_type": "body", "page_no": i + 2, "total_pages": 10,
            "headline": b.get("headline", ""), "body": b.get("body", ""),
            "accent_color": accent, "cta": "",
        })
    pages.append({
        "page_type": "cta", "page_no": 10, "total_pages": 10,
        "headline": cta.get("headline", ""), "body": cta.get("body", ""),
        "accent_color": accent, "cta": cta.get("cta", ""),
    })
    return pages


async def render_slate(slate_id: int, template: str = "default/card.html.j2") -> List[str]:
    async with _render_semaphore():
        return await _render_slate_locked(slate_id, template)


async def _render_slate_locked(slate_id: int, template: str) -> List[str]:
    slate = db.get_card_slate(slate_id)
    if not slate:
        raise ValueError(f"slate {slate_id} not found")
    env = _env()

    # template 파일이 없으면 default로 폴백 (INSTA_DEFAULT_THEME가 import 안 된 theme이면 안전)
    template_full = Path(_resolve_template_dir()) / template
    if not template_full.exists():
        logger.warning("Template '%s' 없음 → 'default/card.html.j2'로 폴백", template)
        template = "default/card.html.j2"

    tmpl = env.get_template(template)
    pages = _build_pages(slate)
    out_dir = _slate_dir(slate_id)
    paths: List[str] = []

    browser = await _get_browser()
    ctx = await browser.new_context(viewport={"width": 1080, "height": 1350})
    try:
        page = await ctx.new_page()
        for spec in pages:
            html_str = tmpl.render(**spec)
            with tempfile.NamedTemporaryFile("w", suffix=".html", delete=False, encoding="utf-8") as f:
                f.write(html_str)
                html_path = f.name
            try:
                await page.goto(f"file://{html_path}", wait_until="networkidle")
                out_path = os.path.join(out_dir, f"{spec['page_no']:02d}.png")
                await page.screenshot(path=out_path, full_page=False, omit_background=False)
                with open(out_path, "rb") as fp:
                    file_hash = hashlib.md5(fp.read()).hexdigest()
                db.add_card_asset(slate_id, spec["page_no"], out_path, file_hash)
                paths.append(out_path)
            finally:
                try:
                    os.unlink(html_path)
                except OSError:
                    pass
    finally:
        await ctx.close()
    return paths