"""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