"""파이프라인 각 Stage 실행 후 자동으로 HTML 시각화를 생성.
save_snapshot()에서 호출됨. 각 stage의 실제 context 데이터로 HTML 생성.
JSON context 파일과 동일한 stage 이름을 사용.
생성되는 파일 (JSON context 파일과 1:1 매칭):
stage_0.html — MDX 정규화 결과 (섹션, 팝업, 이미지)
stage_1a.html — Kei 꼭지 + 영역 배정 (테이블)
stage_1b.html — 컨셉 구체화 (source_data, summary 추가)
stage_1_5a.html — 빈 컨테이너 (1280x720)
stage_1_5a_content.html — 컨테이너에 실제 콘텐츠 배치
stage_1_5b.html — 디자인 예산 (영역별 available_height/width)
stage_1_7.html — 블록 선택 표시
stage_1_8_fit_before.html — 적합성 검증 (재배분 전)
stage_1_8_fit_after.html — 재배분 후 + 보강 결과
stage_1_8_blocks.html — 재배분 후 컨테이너에 블록 배치
stage_2.html — HTML 생성 결과 (영역별 생성된 HTML)
stage_3.html — 렌더링 조립 → final.html 링크
stage_4.html — 품질 게이트 (측정값, 점수)
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from src.pipeline_context import PipelineContext
logger = logging.getLogger(__name__)
COLORS_A = {"배경": "#dc2626", "본심": "#2563eb", "첨부": "#16a34a", "결론": "#7c3aed"}
FONT_MAP_A = {"배경": "bg", "본심": "core", "첨부": "sidebar", "결론": "key_msg"}
# Type B용 동적 색상 팔레트
_COLOR_PALETTE = ["#2563eb", "#16a34a", "#d97706", "#7c3aed", "#dc2626", "#0891b2"]
# 하위 호환: 기존 코드에서 COLORS/FONT_MAP 참조하는 곳 대응
COLORS = COLORS_A
FONT_MAP = FONT_MAP_A
def _is_type_b(ctx) -> bool:
"""page_structure에 zone 키가 있으면 Type B."""
ps = ctx.page_structure.roles if hasattr(ctx.page_structure, 'roles') else {}
for info in ps.values():
if isinstance(info, dict) and info.get("zone") in ("top", "bottom_left", "bottom_right"):
return True
return False
def _get_roles(ctx) -> list[str]:
"""page_structure의 실제 역할명 목록 (순서: zone 기준)."""
ps = ctx.page_structure.roles if hasattr(ctx.page_structure, 'roles') else {}
if _is_type_b(ctx):
zone_order = {"top": 0, "bottom_left": 1, "bottom_right": 2, "footer": 3}
roles = []
for role_name, info in ps.items():
if isinstance(info, dict):
z = info.get("zone", "")
roles.append((zone_order.get(z, 9), role_name))
return [r for _, r in sorted(roles)]
else:
return ["배경", "본심", "첨부", "결론"]
def _get_color(role: str, ctx=None) -> str:
"""역할명 → 색상. Type A는 고정, Type B는 동적."""
if role in COLORS_A:
return COLORS_A[role]
if ctx:
roles = _get_roles(ctx)
idx = roles.index(role) if role in roles else 0
return _COLOR_PALETTE[idx % len(_COLOR_PALETTE)]
return "#666666"
def generate_step_html(stage_name: str, ctx: "PipelineContext", steps_dir: Path) -> None:
"""stage_name에 따라 적절한 시각화 HTML 생성. 파일명은 JSON context와 1:1 매칭."""
try:
if stage_name == "stage_0":
_gen_stage_0(ctx, steps_dir)
elif stage_name == "stage_1a":
_gen_stage_1a(ctx, steps_dir)
elif stage_name == "stage_1b":
_gen_stage_1b(ctx, steps_dir)
elif stage_name == "stage_1_5a":
_gen_stage_1_5a(ctx, steps_dir)
_gen_stage_1_5a_content(ctx, steps_dir)
elif stage_name == "stage_1_5b":
_gen_stage_1_5b(ctx, steps_dir)
elif stage_name == "stage_1_7":
_gen_stage_1_7(ctx, steps_dir)
elif stage_name == "stage_1_8":
# before와 filled는 pipeline.py Stage 1.8 내부에서 before context로 이미 생성됨
# step_visualizer는 after context로 호출되므로 덮어쓰면 안 됨
# blocks와 fit_after만 생성 (after 상태 반영)
_gen_stage_1_8_blocks(ctx, steps_dir)
_gen_stage_1_8_fit_after(ctx, steps_dir)
elif stage_name == "stage_2":
_gen_stage_2(ctx, steps_dir)
elif stage_name == "stage_3":
_gen_stage_3(ctx, steps_dir)
elif stage_name == "stage_4":
_gen_stage_4(ctx, steps_dir)
except Exception as e:
logger.warning(f"[step_viz] {stage_name} 시각화 실패: {e}")
# ══════════════════════════════════════
# 공통
# ══════════════════════════════════════
def _tokens():
from src.fit_verifier import _load_design_tokens
return _load_design_tokens()
def _calc_coords(containers: dict, ratio: tuple, ctx=None) -> dict:
"""역할별 좌표 계산. Type A/B 자동 분기."""
t = _tokens()
pad = t.get("spacing_page", 40)
gap = t.get("spacing_block", 20)
small = t.get("spacing_small", 8)
header_h = 66
inner_w = 1280 - pad * 2
def gh(c):
if hasattr(c, "height_px"): return c.height_px
return c.get("height_px", 0) if isinstance(c, dict) else 0
def gw(c):
if hasattr(c, "width_px"): return c.width_px
return c.get("width_px", 0) if isinstance(c, dict) else 0
# Type B 감지
if ctx and _is_type_b(ctx):
ps = ctx.page_structure.roles
coords = {"header": {"l": pad, "t": pad, "w": inner_w, "h": header_h}}
# zone별 컨테이너 찾기
zone_map = {}
for role_name, info in ps.items():
if isinstance(info, dict):
zone_map[info.get("zone", "")] = role_name
top_role = zone_map.get("top", "")
bl_role = zone_map.get("bottom_left", "")
br_role = zone_map.get("bottom_right", "")
ft_role = zone_map.get("footer", "")
top_h = gh(containers.get(top_role, {}))
bl_h = gh(containers.get(bl_role, {}))
br_h = gh(containers.get(br_role, {}))
ft_h = gh(containers.get(ft_role, {}))
top_top = pad + header_h + gap
bottom_top = top_top + top_h + small
bottom_h = max(bl_h, br_h)
ft_top = bottom_top + bottom_h + gap
bottom_col_w = (inner_w - gap) // 2
if top_role:
coords[top_role] = {"l": pad, "t": top_top, "w": inner_w, "h": top_h}
if bl_role:
coords[bl_role] = {"l": pad, "t": bottom_top, "w": bottom_col_w, "h": bottom_h}
if br_role:
coords[br_role] = {"l": pad + bottom_col_w + gap, "t": bottom_top, "w": bottom_col_w, "h": bottom_h}
if ft_role:
coords[ft_role] = {"l": pad, "t": ft_top, "w": inner_w, "h": ft_h}
return coords
# Type A (기존)
body_w = int(inner_w * ratio[0] / 100) if ratio[0] > 0 else inner_w
sidebar_w = inner_w - body_w - gap if ratio[1] > 0 else 0
bg_h = gh(containers.get("배경", {}))
core_h = gh(containers.get("본심", {}))
sb_h = gh(containers.get("첨부", {}))
ft_h = gh(containers.get("결론", {}))
bg_top = pad + header_h + gap
core_top = bg_top + bg_h + small
ft_top = max(core_top + core_h, bg_top + sb_h) + gap
return {
"header": {"l": pad, "t": pad, "w": inner_w, "h": header_h},
"배경": {"l": pad, "t": bg_top, "w": body_w, "h": bg_h},
"본심": {"l": pad, "t": core_top, "w": body_w, "h": core_h},
"첨부": {"l": pad + body_w + gap, "t": bg_top, "w": sidebar_w, "h": sb_h},
"결론": {"l": pad, "t": ft_top, "w": inner_w, "h": ft_h},
}
def _wrap(title, subtitle, slide_body, ctx=None):
"""slide-base.html 기반 래핑. step 시각화도 실제 슬라이드와 같은 기반 사용."""
from pathlib import Path
slide_base_path = Path(__file__).parent.parent / "templates" / "blocks" / "slide-base.html"
slide_title = ""
footer_text = ""
if ctx:
slide_title = ctx.analysis.title if ctx.analysis else ""
footer_text = ctx.analysis.conclusion_text if ctx.analysis else ""
try:
raw = slide_base_path.read_text(encoding="utf-8")
# {% block body %} → slide_body로 치환
raw = raw.replace("{% block body %}{% endblock %}", slide_body)
from jinja2 import Template
template = Template(raw)
slide_html = template.render(title=slide_title, footer_text=footer_text, footer_pill_bg="")
# step 라벨 추가
label = (f''
f'{title}
'
f''
f'{subtitle}
')
return slide_html.replace('', f'{label}')
except Exception:
# fallback: 기존 방식
return f"""
{title}
{subtitle}
{slide_body}
"""
def _hdr(c, title):
return (f'{title}
\n')
def _box(c, role, inner, extra=""):
cl = COLORS.get(role, "#333")
return (f'{inner}
\n')
# ══════════════════════════════════════
# Stage 0: MDX 정규화
# ══════════════════════════════════════
def _gen_stage_0(ctx, steps_dir):
"""MDX 정규화 결과: 섹션, 팝업, 이미지, 테이블 목록."""
norm = ctx.normalized if hasattr(ctx, 'normalized') else {}
if hasattr(norm, 'model_dump'):
norm = norm.model_dump()
elif not isinstance(norm, dict):
norm = {}
sections = norm.get("sections", [])
popups = norm.get("popups", [])
images = norm.get("images", [])
tables = norm.get("tables", [])
title = norm.get("title", ctx.analysis.title if ctx.analysis else "")
sec_rows = ""
for i, s in enumerate(sections):
heading = s.get("heading", "") if isinstance(s, dict) else ""
content = s.get("content", "") if isinstance(s, dict) else str(s)
preview = content[:120].replace("<", "<") + ("..." if len(content) > 120 else "")
bg = "#f8fafc" if i % 2 == 0 else "#fff"
sec_rows += f'| {i+1} | {heading} | {preview} |
\n'
popup_rows = ""
for p in popups:
pt = p.get("title", "") if isinstance(p, dict) else str(p)
pc = p.get("content", "") if isinstance(p, dict) else ""
popup_rows += f'| {pt} | {len(pc)}자 |
\n'
html = f"""
Stage 0: MDX 정규화
제목: {title} | 섹션: {len(sections)}개 | 팝업: {len(popups)}개 | 이미지: {len(images)}개 | 테이블: {len(tables)}개
섹션
| # | heading | content (미리보기) |
|---|
{sec_rows}
팝업
"""
(steps_dir / "stage_0.html").write_text(html, encoding="utf-8")
# ══════════════════════════════════════
# Stage 1A: Kei 꼭지
# ══════════════════════════════════════
def _gen_stage_1a(ctx, steps_dir):
ps = ctx.page_structure.roles
rm = {}
for role, info in ps.items():
if isinstance(info, dict):
for tid in info.get("topic_ids", []):
rm[tid] = role
rows = ""
for t in ctx.topics:
role = rm.get(t.id, "?")
c = COLORS.get(role, "#333")
bg = "#f8fafc" if t.id % 2 == 0 else "#fff"
rows += (f'| {t.id} | '
f'{t.title} | '
f'{t.purpose} | {t.layer} | '
f'{t.relation_type} | '
f'{role} |
\n')
ps_info = "
".join(f"{r}: topic_ids={info.get('topic_ids')}, weight={info.get('weight')}"
for r, info in ps.items() if isinstance(info, dict))
html = f"""
Stage 1A/1B: Kei 꼭지 + 영역 배정
| ID | 제목 |
purpose | layer | relation_type |
영역 |
|---|
{rows}
페이지 구조:
{ps_info}
"""
(steps_dir / "stage_1a.html").write_text(html, encoding="utf-8")
# ══════════════════════════════════════
# Stage 1B: 컨셉 구체화
# ══════════════════════════════════════
def _gen_stage_1b(ctx, steps_dir):
"""Stage 1B 후 꼭지에 source_data, summary가 추가된 상태."""
ps = ctx.page_structure.roles
rm = {}
for role, info in ps.items():
if isinstance(info, dict):
for tid in info.get("topic_ids", []):
rm[tid] = role
rows = ""
for t in ctx.topics:
role = rm.get(t.id, "?")
c = COLORS.get(role, "#333")
bg = "#f8fafc" if t.id % 2 == 0 else "#fff"
sd = (t.source_data or "")[:150]
sd_display = sd.replace("<", "<") + ("..." if len(t.source_data or "") > 150 else "")
summary = (t.summary or "")[:100] if hasattr(t, 'summary') else ""
rows += (f'| {t.id} | '
f'{t.title} | '
f'{role} | '
f'{t.layer} | '
f'{sd_display} | '
f'{summary} |
\n')
html = f"""
Stage 1B: 컨셉 구체화
Stage 1A의 꼭지에 source_data(원본 텍스트)와 summary가 추가됨
| ID | 제목 |
영역 | layer |
source_data (미리보기) | summary |
|---|
{rows}
"""
(steps_dir / "stage_1b.html").write_text(html, encoding="utf-8")
# ══════════════════════════════════════
# Stage 1.5a: 빈 컨테이너
# ══════════════════════════════════════
def _gen_stage_1_5a(ctx, steps_dir):
"""slide-base 위에 빈 zone 컨테이너만 표시."""
ps = ctx.page_structure.roles
gap = 8
# zone 순서
zone_priority = {"top": 0, "bottom": 1, "bottom_left": 2, "bottom_right": 3}
roles_sorted = sorted(
[(r, info) for r, info in ps.items() if isinstance(info, dict)],
key=lambda x: zone_priority.get(x[1].get("zone", ""), 9),
)
body_html = ""
for i, (role, info) in enumerate(roles_sorted):
ci = ctx.containers.get(role)
if not ci:
continue
cl = _get_color(role, ctx)
zone = info.get("zone", "")
w = ci.width_px
h = ci.height_px
tids = info.get("topic_ids", [])
body_html += (
f''
f'
'
f'{role}
'
f'zone: {zone} / {w}×{h}px
'
f'topics: {tids}'
f'
{role} ({w}×{h}px)
']
for tid in tids:
t = topic_map.get(tid)
if not t:
continue
lines.append(f'[꼭지{tid}] {t.title} — {t.purpose} · {t.layer}
')
sd = t.source_data
if sd:
for sent in sd.split(", ")[:5]:
sent = sent.strip()
if sent:
lines.append(f'• {sent}
')
body_html += (
f''
f'{"".join(lines)}
\n'
)
html = _wrap("Stage 1.5a: 콘텐츠 배치", "zone별 topic source_data", body_html, ctx=ctx)
(steps_dir / "stage_1_5a_content.html").write_text(html, encoding="utf-8")
# ══════════════════════════════════════
# Stage 1.5b: 디자인 예산
# ══════════════════════════════════════
def _gen_stage_1_5b(ctx, steps_dir):
"""slide-base 위 zone별 디자인 예산."""
ps = ctx.page_structure.roles
gap = 8
zone_priority = {"top": 0, "bottom": 1, "bottom_left": 2, "bottom_right": 3}
roles_sorted = sorted(
[(r, info) for r, info in ps.items() if isinstance(info, dict)],
key=lambda x: zone_priority.get(x[1].get("zone", ""), 9),
)
body_html = ""
for role, info in roles_sorted:
ci = ctx.containers.get(role)
if not ci:
continue
cl = _get_color(role, ctx)
w, h = ci.width_px, ci.height_px
db = ci.design_budget
if db and hasattr(db, 'model_dump'):
db = db.model_dump()
elif not isinstance(db, dict):
db = {}
avail_h = db.get("available_height_px", 0)
avail_w = db.get("available_width_px", 0)
fits = db.get("fits", False)
icon = "✅" if fits else "⚠️"
body_html += (
f''
f'
{icon} {role} ({w}×{h}px)
'
f'
available: {avail_h}×{avail_w}px / fits: {fits}
'
f'
{role} ({w}×{h}px)
']
for r in ref_list:
vtype_label = "tag_match ✅" if r.visual_type == "tag_match" else r.visual_type
lines.append(f'{r.block_id} ({r.variant}) — {vtype_label}
')
body_html += (
f''
f'{"".join(lines)}
\n'
)
html = _wrap("Stage 1.7: 블록 선택", "tag 매칭 기반 블록 선택", body_html, ctx=ctx)
(steps_dir / "stage_1_7.html").write_text(html, encoding="utf-8")
# ══════════════════════════════════════
# Stage 1.8: 텍스트/그림 채운 상태 (filled)
# ══════════════════════════════════════
def _gen_stage_1_8_filled(ctx, steps_dir):
"""블록 디자인에 텍스트를 채운 상태. 공통 조립 함수(block_assembler) 사용."""
from src.block_assembler import assemble_slide_html
slide_html = assemble_slide_html(ctx)
# 시각화 제목 삽입
header = (
''
'Stage 1.8: 블록 디자인에 텍스트 채운 상태 (fit 검증 전)
\n'
''
'블록 CSS + structured_text + font_hierarchy. 공통 조립 함수 사용.
\n'
)
html = slide_html.replace('', '\n' + header, 1)
(steps_dir / "stage_1_8_filled.html").write_text(html, encoding="utf-8")
def _gen_stage_1_8_fit_before(ctx, steps_dir):
"""slide-base 위 zone별 초기 배정 (weight 기반)."""
ps = ctx.page_structure.roles
gap = 8
zone_priority = {"top": 0, "bottom": 1, "bottom_left": 2, "bottom_right": 3}
roles_sorted = sorted(
[(r, info) for r, info in ps.items() if isinstance(info, dict)],
key=lambda x: zone_priority.get(x[1].get("zone", ""), 9),
)
body_html = ""
for role, info in roles_sorted:
ci = ctx.containers.get(role)
if not ci:
continue
cl = _get_color(role, ctx)
w, h = ci.width_px, ci.height_px
weight = info.get("weight", 0)
ref_list = ctx.references.get(role, [])
blocks = ", ".join(r.block_id for r in ref_list) if ref_list else "미선택"
body_html += (
f''
f'
{role} ({w}×{h}px)
'
f'
weight: {weight} / 블록: {blocks}
'
f'
'
f'
{icon} {role} ({w}×{new_h}px){delta_str}
'
f'
블록: {blocks}
'
f'
Stage 2: slide-base + 블록 템플릿 조립 결과 (Type B)
slide-base.html 배경 위에 블록 템플릿으로 조립된 최종 HTML
"""
(steps_dir / "stage_2.html").write_text(html, encoding="utf-8")
return
# Type A: dict (body_html, sidebar_html, footer_html)
import re as _re
body_html = gen.get("body_html", "")
sidebar_html = gen.get("sidebar_html", "")
footer_html = gen.get("footer_html", "")
spacer_pattern = r''
body_parts = _re.split(spacer_pattern, body_html, maxsplit=1)
bg_html = body_parts[0].strip() if len(body_parts) > 1 else ""
core_html = body_parts[1].strip() if len(body_parts) > 1 else body_html.strip()
role_htmls = {}
if bg_html and "배경" in ps:
role_htmls["배경"] = bg_html
if core_html and "본심" in ps:
role_htmls["본심"] = core_html
if sidebar_html and "첨부" in ps:
role_htmls["첨부"] = sidebar_html
if footer_html and "결론" in ps:
role_htmls["결론"] = footer_html
# 각 역할을 컨테이너 크기에 맞게 실제 렌더링
fit = ctx.fit_result or {}
redist = fit.get("redistribution", {})
sections = []
for role in _get_roles(ctx):
rhtml = role_htmls.get(role, "")
if not rhtml:
continue
cl = _get_color(role, ctx)
ci = ctx.containers.get(role)
if not ci:
continue
h = int(redist.get(role, ci.height_px))
w = ci.width_px
# sub_layout 정보
layout = sub_layouts.get(role, {})
scs = layout.get("sub_containers", [])
sc_desc = " + ".join(f'{sc["name"]}({int(sc["width_px"])}×{int(sc["height_px"])}px)' for sc in scs) if scs else ""
sections.append(
f''
f'
'
f'{role} ({w}×{h}px)'
f'{" — " + sc_desc if sc_desc else ""}
'
f'
{rhtml}
Stage 2: 영역별 HTML 생성 결과 (Sonnet)
각 역할의 Sonnet 출력을 컨테이너 크기에 맞게 실제 렌더링
{"".join(sections)}
"""
(steps_dir / "stage_2.html").write_text(html, encoding="utf-8")
def _gen_stage_3(ctx, steps_dir):
"""Stage 3 결과: rendered_html을 별도 파일로 저장 + 링크.
rendered_html 자체가 완성 HTML이므로 직접 렌더링 가능."""
rendered = ctx.rendered_html or ""
if rendered:
# rendered_html을 별도 파일로 저장하여 브라우저에서 직접 확인 가능
(steps_dir / "stage_3_rendered.html").write_text(rendered, encoding="utf-8")
html = f"""
Stage 3: 렌더링 조립 결과
Stage 2의 영역별 HTML을 슬라이드 프레임(CSS Grid)에 배치 + 후처리 적용
렌더링 결과 보기 (1280×720) →
final.html 보기 →
Stage 3 후처리: sidebar width:100% 조정, 폰트 캡핑 (배경≤{ctx.font_hierarchy.bg}px, 첨부≤{ctx.font_hierarchy.sidebar}px), overflow 제거, bold 변환
"""
(steps_dir / "stage_3.html").write_text(html, encoding="utf-8")
# ══════════════════════════════════════
# Stage 4: 품질 게이트
# ══════════════════════════════════════
def _gen_structure_validation(ctx) -> str:
"""sample-based 구조 검증. "어긋나면 안 된다" 기준."""
import re as _re
checks = []
html = ctx.rendered_html if hasattr(ctx, 'rendered_html') else ""
ps = ctx.page_structure.roles if hasattr(ctx.page_structure, 'roles') else {}
# 1. 본문 텍스트 visible (body에 실제 텍스트가 있는지)
body_start = html.find(']+>', '', html[body_start:]) if body_start > 0 else ""
body_text = _re.sub(r'\s+', ' ', body_text).strip()
text_len = len(body_text)
ok = text_len > 100
checks.append(("본문 텍스트 visible", f"{'✅' if ok else '❌'} {text_len}자"))
# 2. detail link 개수 (role당 1개)
link_count = len(_re.findall(r'자세히보기', html)) if html else 0
popup_count = len(ctx.normalized.popups) if hasattr(ctx.normalized, 'popups') else 0
ok = link_count <= max(popup_count, 1)
checks.append(("detail link 개수", f"{'✅' if ok else '⚠️'} {link_count}개 (popup {popup_count}개)"))
# 3. body 안
Stage 4: 품질 게이트
품질 점수: {quality_score}
슬라이드: clientHeight={slide_m.get("clientHeight", "?")}px, scrollHeight={slide_m.get("scrollHeight", "?")}px, overflow={slide_m.get("overflowed", "?")}
Overflow 측정
| 영역 | clientH | scrollH | excess |
|---|
{zone_rows}
블록/Recipe 선택
| zone | role | schema | block/recipe |
|---|
{recipe_rows}
Popup 연결
| popup_id | target_role | popup_file |
|---|
{popup_rows if popup_rows else '| 없음 |
'}
구조 검증
| 검증 항목 | 결과 |
|---|
{_gen_structure_validation(ctx)}
"""
(steps_dir / "stage_4.html").write_text(html, encoding="utf-8")