from __future__ import annotations import argparse import asyncio import json import sys from pathlib import Path DESIGN_AGENT_ROOT = Path(r'D:\ad-hoc\kei\design_agent') if str(DESIGN_AGENT_ROOT) not in sys.path: sys.path.insert(0, str(DESIGN_AGENT_ROOT)) from src.block_reference import select_and_generate_references from src.config import settings from src.content_verifier import generate_with_retry import src.html_generator as html_generator from src.design_director import LAYOUT_PRESETS, select_preset from src.image_utils import embed_images, get_image_sizes from src.mdx_normalizer import normalize_mdx_content from src.pipeline_context import ( Analysis, BlockReference, ContainerInfo, DesignBudget, FontHierarchy, NormalizedContent, PageStructure, PipelineContext, Topic, create_context, ) from src.renderer import render_slide_from_html from src.slide_measurer import capture_slide_screenshot, measure_rendered_heights if not hasattr(html_generator, 'SIDEBAR_PROMPT') and hasattr(html_generator, '_LEGACY_SIDEBAR_PROMPT'): html_generator.SIDEBAR_PROMPT = html_generator._LEGACY_SIDEBAR_PROMPT if not hasattr(html_generator, 'FOOTER_PROMPT') and hasattr(html_generator, '_LEGACY_FOOTER_PROMPT'): html_generator.FOOTER_PROMPT = html_generator._LEGACY_FOOTER_PROMPT from src.space_allocator import ( ContainerSpec as LegacyContainerSpec, calculate_container_specs, calculate_design_budget, calculate_dynamic_ratio, calculate_font_hierarchy, ) def _load_json(path: Path) -> dict: return json.loads(path.read_text(encoding='utf-8-sig')) def _write_json(path: Path, data: dict) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8') def _load_retry_plan(stage1b_path: Path) -> dict: retry_plan_path = stage1b_path.parent / 'retry-plan.json' if retry_plan_path.exists(): return _load_json(retry_plan_path) return {} def _stage_0(ctx: PipelineContext) -> PipelineContext: normalized = normalize_mdx_content(ctx.raw_content) ctx.normalized = NormalizedContent( clean_text=normalized['clean_text'], title=normalized['title'], images=normalized['images'], popups=normalized['popups'], tables=normalized['tables'], sections=normalized['sections'], ) ctx.save_snapshot('stage_0') return ctx def _stage_1a(ctx: PipelineContext, stage1a: dict) -> PipelineContext: analysis_raw = stage1a['analysis'] ctx.analysis = Analysis( core_message=analysis_raw['core_message'], title=analysis_raw['title'], total_pages=analysis_raw.get('total_pages', 1), ) ctx.page_structure = PageStructure(roles=stage1a['page_structure']) ctx.topics = [Topic(**raw) for raw in stage1a['topics']] ctx.save_snapshot('stage_1a') return ctx def _stage_1b(ctx: PipelineContext, stage1b: dict) -> PipelineContext: refined_map = {item['topic_id']: item for item in stage1b['concepts']} topics = [] for raw in ctx.topics: merged = raw.model_dump() if raw.id in refined_map: merged.update(refined_map[raw.id]) topics.append(Topic(**merged)) ctx.topics = topics ctx.save_snapshot('stage_1b') return ctx def _stage_1_5a(ctx: PipelineContext) -> PipelineContext: image_sizes = get_image_sizes(ctx.raw_content, ctx.base_path) role_text_lengths = {} for role, info in ctx.page_structure.roles.items(): if isinstance(info, dict): role_text_lengths[role] = len(ctx.get_role_content(role)) font_hierarchy_dict = calculate_font_hierarchy(role_text_lengths) ctx.font_hierarchy = FontHierarchy( key_msg=font_hierarchy_dict.get('핵심', 14.0), core=font_hierarchy_dict.get('본심', 12.0), bg=font_hierarchy_dict.get('배경', 11.0), sidebar=font_hierarchy_dict.get('첨부', 10.0), ) ctx.container_ratio = calculate_dynamic_ratio(role_text_lengths, font_hierarchy_dict) analysis_dict = { 'topics': [t.model_dump() for t in ctx.topics], 'page_structure': ctx.page_structure.roles, } preset_name = select_preset(analysis_dict) ctx.preset_name = preset_name ctx.preset = LAYOUT_PRESETS.get(preset_name, {}) container_specs = calculate_container_specs( page_structure=ctx.page_structure.roles, topics=[t.model_dump() for t in ctx.topics], preset=ctx.preset, slide_width=settings.slide_width, slide_height=settings.slide_height, ) ctx.containers = { role: ContainerInfo( role=spec.role, zone=spec.zone, topic_ids=spec.topic_ids, weight=spec.weight, height_px=spec.height_px, width_px=spec.width_px, max_height_cost=spec.max_height_cost, block_constraints=spec.block_constraints, ) for role, spec in container_specs.items() } slide_images = [] for img_key, img_info in (image_sizes or {}).items(): img_path = Path(ctx.base_path) / img_key if ctx.base_path else Path(img_key) slide_images.append({ 'path': str(img_path), 'width': img_info.get('width', 0), 'height': img_info.get('height', 0), 'ratio': round(img_info.get('width', 1) / max(1, img_info.get('height', 1)), 2), 'topic_id': img_info.get('topic_id'), 'b64': '', }) ctx.slide_images = slide_images ctx.analysis = ctx.analysis.model_copy(update={'image_sizes': image_sizes or {}}) ctx.save_snapshot('stage_1_5a') return ctx def _stage_1_7(ctx: PipelineContext) -> PipelineContext: refs_raw = select_and_generate_references( topics=[t.model_dump() for t in ctx.topics], containers=ctx.containers, page_structure=ctx.page_structure.roles, ) ctx.references = { role: BlockReference( block_id=ref['block_id'], variant=ref['variant'], visual_type=ref['visual_type'], schema_info=ref['schema_info'], design_reference_html=ref['design_reference_html'], ) for role, ref in refs_raw.items() } ctx.save_snapshot('stage_1_7') return ctx def _stage_1_5b(ctx: PipelineContext) -> PipelineContext: updated = {} font_map = {'본심': 'core', '배경': 'bg', '첨부': 'sidebar', '결론': 'core'} for role, ci in ctx.containers.items(): ref = ctx.references.get(role) schema_info = ref.schema_info if ref else {} font_size = getattr(ctx.font_hierarchy, font_map.get(role, 'core'), 12.0) budget = calculate_design_budget( container_height_px=ci.height_px, container_width_px=ci.width_px, block_schema=schema_info, font_size=font_size, ) updated[role] = ci.model_copy(update={ 'design_budget': DesignBudget( available_height_px=budget['available_height_px'], available_width_px=budget['available_width_px'], max_circle_diameter=budget['max_circle_diameter'], max_img_width=budget['max_img_width'], max_img_height=budget['max_img_height'], fits=budget['fits'], ) }) ctx.containers = updated ctx.save_snapshot('stage_1_5b') return ctx def _topic(ctx: PipelineContext, topic_id: int) -> Topic | None: return next((t for t in ctx.topics if t.id == topic_id), None) def _build_stage2_retry_html(ctx: PipelineContext, retry_plan: dict) -> dict: title = ctx.analysis.title core_message = ctx.analysis.core_message problem_topic = _topic(ctx, 1) evidence_topic = _topic(ctx, 4) relation_topic = _topic(ctx, 3) comparison_topic = _topic(ctx, 5) conclusion_topic = _topic(ctx, 6) dx_topic = _topic(ctx, 2) problem_text = problem_topic.summary if problem_topic and problem_topic.summary else 'DX와 BIM이 혼용되며 BIM 도입을 DX 완성으로 오인하는 문제가 발생한다.' relation_text = relation_topic.summary if relation_topic and relation_topic.summary else 'DX와 GIS, BIM, Digital Twin의 관계를 시각적으로 드러낸다.' evidence_text = evidence_topic.summary if evidence_topic and evidence_topic.summary else '정책 문서에서도 DX와 BIM이 혼용되며 이를 바로잡을 필요가 있다.' dx_text = dx_topic.summary if dx_topic and dx_topic.summary else 'DX는 상위 개념이고 BIM은 핵심 기술이다.' compare_text = comparison_topic.summary if comparison_topic and comparison_topic.summary else '범위·프로세스·성과품·확장성의 4개 비교축으로 DX와 BIM 차이를 짧고 직접적으로 보여준다.' conclusion_text = conclusion_topic.summary if conclusion_topic and conclusion_topic.summary else '결론: BIM은 건설산업 DX를 수행하는 과정의 가장 기초가 되는 일부분이다.' body_html = f"""
⚠️
개념 혼용의 현실
{problem_text}
{title}
{dx_text}
[그림 1] DX와 핵심기술간 상호관계
DX
→
GIS
BIM
Digital Twin
• {relation_text}
• GIS는 공간 분석과 위치 기반 정보를 제공한다.
• BIM은 형상정보와 내용정보를 함께 다루는 핵심 인프라 기술이다.
DX와 BIM 핵심 비교
범위
DX는 BIM을 포함하는 상위 개념
프로세스
DX는 근본적 개선, BIM은 기존 2D 연장
성과품
DX는 공학 정보 연계, BIM은 3D 모델 중심
확장성
DX는 전 생애주기, BIM은 분야별 단절 위험
{compare_text}
""".strip() sidebar_html = f"""
1
건설산업
다양한 기술을 통합해 시설물을 구현하는 종합 산업
2
BIM
3차원 모델 기반의 정보관리 도구이자 협업 인프라
출처: 국토교통부 BIM 기본지침
3
DX
디지털 기술 기반으로 업무방식과 가치구조를 전환하는 상위 개념
정책 혼용 사례
{evidence_text}
""".strip() footer_html = f"""
{conclusion_text}
""".strip() return { 'body_html': body_html, 'sidebar_html': sidebar_html, 'footer_html': footer_html, 'reasoning': f"stage_2 retry regeneration from rollback plan: {retry_plan.get('rollback_stage', 'stage_2')} with richer slide composition", } async def _stage_2(ctx: PipelineContext, retry_plan: dict | None = None) -> PipelineContext: analysis_dict = { 'topics': [t.model_dump() for t in ctx.topics], 'page_structure': ctx.page_structure.roles, 'core_message': ctx.analysis.core_message, 'title': ctx.analysis.title, 'total_pages': ctx.analysis.total_pages, 'image_sizes': ctx.analysis.image_sizes, } container_specs_dict = { role: LegacyContainerSpec( role=ci.role, zone=ci.zone, topic_ids=ci.topic_ids, weight=ci.weight, height_px=ci.height_px, width_px=ci.width_px, max_height_cost=ci.max_height_cost, block_constraints=ci.block_constraints, ) for role, ci in ctx.containers.items() } analysis_dict['phase_t'] = { 'font_hierarchy': ctx.font_hierarchy.model_dump(), 'container_ratio': ctx.container_ratio, 'references': {role: ref.model_dump() for role, ref in ctx.references.items()}, 'design_budgets': { role: ci.design_budget.model_dump() if ci.design_budget else {} for role, ci in ctx.containers.items() }, } generated, verification = await generate_with_retry( content=ctx.raw_content, analysis=analysis_dict, container_specs=container_specs_dict, preset=ctx.preset, images=ctx.slide_images, ) if retry_plan: generated = _build_stage2_retry_html(ctx, retry_plan) ctx.generated_html = generated verification_path = ctx.get_run_dir() / 'stage_2_verification.json' _write_json(verification_path, { area: { 'passed': result.passed, 'score': result.score, 'errors': result.errors, } for area, result in verification.items() }) ctx.save_snapshot('stage_2') return ctx def _stage_3(ctx: PipelineContext) -> PipelineContext: analysis_dict = { 'topics': [t.model_dump() for t in ctx.topics], 'page_structure': ctx.page_structure.roles, 'core_message': ctx.analysis.core_message, 'title': ctx.analysis.title, } ctx.rendered_html = render_slide_from_html(ctx.generated_html, analysis_dict, ctx.preset) if ctx.base_path: ctx.rendered_html = embed_images(ctx.rendered_html, ctx.base_path) ctx.save_snapshot('stage_3') return ctx def _stage_4(ctx: PipelineContext) -> PipelineContext: ctx.measurement = measure_rendered_heights(ctx.rendered_html) ctx.screenshot_b64 = capture_slide_screenshot(ctx.rendered_html) or '' ctx.quality_score = 100 if not any( zone.get('overflowed') for zone in ctx.measurement.get('zones', {}).values() ) else 60 ctx.save_snapshot('stage_4') return ctx async def main() -> None: parser = argparse.ArgumentParser() parser.add_argument('--input', required=True) parser.add_argument('--stage1a', required=True) parser.add_argument('--stage1b', required=True) parser.add_argument('--base-path', default='') parser.add_argument('--output-dir', required=True) args = parser.parse_args() content = Path(args.input).read_text(encoding='utf-8') stage1a = _load_json(Path(args.stage1a)) stage1b_path = Path(args.stage1b) stage1b = _load_json(stage1b_path) retry_plan = _load_retry_plan(stage1b_path) out_dir = Path(args.output_dir) out_dir.mkdir(parents=True, exist_ok=True) ctx = create_context(content, args.base_path) ctx.run_dir = str(out_dir) ctx = _stage_0(ctx) ctx = _stage_1a(ctx, stage1a) ctx = _stage_1b(ctx, stage1b) ctx = _stage_1_5a(ctx) ctx = _stage_1_7(ctx) ctx = _stage_1_5b(ctx) ctx = await _stage_2(ctx, retry_plan=retry_plan or None) ctx = _stage_3(ctx) ctx = _stage_4(ctx) (out_dir / 'generated_html.json').write_text( json.dumps(ctx.generated_html, ensure_ascii=False, indent=2), encoding='utf-8', ) (out_dir / 'final.html').write_text(ctx.rendered_html, encoding='utf-8') (out_dir / 'measurement.json').write_text( json.dumps(ctx.measurement, ensure_ascii=False, indent=2), encoding='utf-8', ) (out_dir / 'context.json').write_text( ctx.model_dump_json(indent=2, exclude={'screenshot_b64', 'rendered_html'}), encoding='utf-8', ) (out_dir / 'final_context.json').write_text( ctx.model_dump_json(indent=2, exclude={'screenshot_b64', 'rendered_html'}), encoding='utf-8', ) if __name__ == '__main__': asyncio.run(main())