KyeongminandClaude Opus 4.6
1f7579cf64
Phase W: - weight 비율 초기 배정 (space_allocator header 높이 반영) - block_assembler 공통 조립 함수 (filled/assembled 통합) - filled → Selenium 측정 → context 저장 - sidebar overflow 확장 + body 재배분 - sub_layouts 사전 계산 (이미지 누락 해결) Phase V': - 팝업 링크 우측상단 배치 (인라인 → position:absolute) - 표 내용 Kei 판단 (공란 크기 계산 → 행/열 산출 → Kei 요약) - 출처 라벨 삭제 + 이미지 아래 캡션 배치 - after 공란 제거 (결론 바로 위까지 body/sidebar 채움) 추가: - V-10 bold 키워드: 기계적 추출 → Kei 문맥 판단 - ** 마크다운 → <strong> 변환 - [이미지:] 마커 제거 (bold 변환 전 처리) - grid-template-rows AFTER 크기 반영 (Sonnet final) - assemble_stage2 CSS font-size override, white-space fix - 하드코딩 전수 검토 완료 - 본심 여러 topic 텍스트 합침 Phase X 계획 문서 작성 (동적 역할 구조) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
476 lines
21 KiB
Python
476 lines
21 KiB
Python
"""DA-13b: 3단계 — Kei 텍스트 편집자 (텍스트 정리).
|
|
|
|
디자인 팀장의 레이아웃 컨셉 + 원본 콘텐츠를 받아,
|
|
Kei API를 통해 도메인 전문가로서 각 슬롯 텍스트를 정리한다.
|
|
팀장의 글자 수 가이드를 참고하되 내용 의미가 우선.
|
|
|
|
Kei API 필수. fallback 없음. 성공할 때까지 무한 재시도.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
import anthropic
|
|
import httpx
|
|
|
|
from src.config import settings
|
|
from src.design_director import BLOCK_SLOTS
|
|
from src.sse_utils import stream_sse_tokens
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
EDITOR_PROMPT = """당신은 도메인 전문가이자 콘텐츠 편집자이다.
|
|
원본 콘텐츠의 핵심 내용을 유지하면서 각 블록의 슬롯에 맞게 텍스트를 정리한다.
|
|
|
|
## 핵심 원칙
|
|
- **원본 텍스트를 최대한 보존한다.** 슬라이드 공간에 맞게 약간만 축약한다.
|
|
- 의미를 바꾸거나 완전히 재작성하지 않는다.
|
|
- 글자수 예산(★ 표시)이 있으면 반드시 지킨다. 초과하면 overflow가 발생한다.
|
|
- 예산 내라면 원본을 최대한 보존. 예산 초과 시에만 뒤에서부터 축약.
|
|
- 디자인 실무자가 텍스트에 맞게 디자인을 조정할 것이므로, 텍스트를 억지로 자르지 않는다.
|
|
- **모든 슬롯을 빠짐없이 채운다. 빈 슬롯 금지.**
|
|
|
|
## 편집 규칙
|
|
- 전체 컨텍스트와 핵심 용어를 보존한다
|
|
- 개조식(불릿, 번호)으로 작성한다. 줄글 금지.
|
|
- 각 블록의 **목적(purpose)**을 보고 해당 목적에 맞는 텍스트를 원본에서 가져온다
|
|
- **불릿 항목은 반드시 각각 별도 줄(\n)로 작성한다.** 한 줄에 여러 항목을 넣지 마라.
|
|
- 올바른 예: "• 추진과제: 건설산업 디지털화\n• 실행과제: BIM 전면 도입"
|
|
- 잘못된 예: "• 추진과제: 건설산업 디지털화 • 실행과제: BIM 전면 도입"
|
|
- 출처가 있는 내용은 출처를 반드시 보존한다
|
|
- 출처가 없는 수치나 통계를 만들지 않는다
|
|
|
|
## 표 편집 규칙
|
|
- 표는 표로 유지한다 (다른 형태로 전환하지 않음)
|
|
- 팀장이 요약 요청하면 핵심 행/열만 선택하고 "...외 N건" 표기
|
|
|
|
## 자세히보기 편집 규칙
|
|
- detail_target인 꼭지는 두 버전을 작성:
|
|
- summary: 슬라이드 표면에 보일 요약 (3줄 이내)
|
|
- detail: 펼치면 보일 전체 내용
|
|
|
|
## 분량 원칙
|
|
- 각 블록의 ★ 컨테이너 제약을 확인하고 그 범위 안에서 작성한다.
|
|
- 컨테이너 제약이 없으면 원본 텍스트를 최대한 보존한다.
|
|
- 비교 블록 사용 시: 비교 목적(왜 비교하는가)을 첫 행 또는 상단에 요약.
|
|
|
|
## source 슬롯 규칙 (절대 규칙)
|
|
- source 슬롯에는 반드시 정보원(출처)을 넣는다
|
|
- 꼭지 제목, 주제어, 섹션명을 source에 넣지 마라
|
|
- 출처가 원본에 없으면 source 슬롯을 비워라 (빈 문자열)
|
|
- 올바른 예: '국토교통부, 2020', 'IBM, 2011'
|
|
- 잘못된 예: '용어의 혼용', 'DX와 BIM 개념'
|
|
|
|
## JSON 형식으로만 응답한다. 설명 없이 JSON만."""
|
|
|
|
|
|
async def fill_content(
|
|
content: str,
|
|
layout_concept: dict[str, Any],
|
|
analysis: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""3단계: 각 페이지의 각 블록 슬롯에 텍스트를 채운다.
|
|
|
|
Args:
|
|
content: 원본 텍스트 콘텐츠
|
|
layout_concept: 디자인 팀장의 레이아웃 컨셉
|
|
analysis: 1단계 실장의 꼭지 분석 결과 (참고용)
|
|
|
|
Returns:
|
|
슬롯이 채워진 layout_concept
|
|
"""
|
|
for page_idx, page in enumerate(layout_concept.get("pages", [])):
|
|
blocks = page.get("blocks", [])
|
|
if not blocks:
|
|
continue
|
|
|
|
# 블록별 슬롯 + 글자 수 가이드 생성
|
|
slot_requirements = []
|
|
for i, block in enumerate(blocks):
|
|
block_type = block.get("type", "")
|
|
slots = BLOCK_SLOTS.get(block_type, {})
|
|
char_guide = block.get("char_guide", {})
|
|
|
|
topic_id = block.get("topic_id", i + 1)
|
|
|
|
# Phase Q: topic의 source_data를 찾아서 직접 전달
|
|
source_data_text = ""
|
|
if analysis:
|
|
for topic in analysis.get("topics", []):
|
|
if topic.get("id") == topic_id:
|
|
sd = topic.get("source_data", "")
|
|
if sd:
|
|
source_data_text = sd
|
|
break
|
|
|
|
req_text = (
|
|
f"블록 {i+1} ({block_type}, 영역: {block.get('area', '?')}, topic_id: {topic_id}):\n"
|
|
f" 목적(purpose): {block.get('purpose', '미지정')}\n"
|
|
f" 필수 슬롯: {slots.get('required', [])}\n"
|
|
f" 선택 슬롯: {slots.get('optional', [])}"
|
|
)
|
|
|
|
# source_data를 최우선으로 전달
|
|
if source_data_text:
|
|
req_text += (
|
|
f"\n ★★ source_data (이 텍스트를 그대로 슬롯에 배치하라):\n"
|
|
f" {source_data_text}"
|
|
)
|
|
|
|
# I-5: 슬롯 의미 설명 전달 (slot_desc가 있으면)
|
|
slot_desc = slots.get("slot_desc", {})
|
|
if slot_desc:
|
|
desc_lines = [f" {k}: {v}" for k, v in slot_desc.items()]
|
|
req_text += "\n 슬롯 설명:\n" + "\n".join(desc_lines)
|
|
|
|
if char_guide:
|
|
guide_lines = [f" {k}: ~{v}자" for k, v in char_guide.items()]
|
|
req_text += "\n 글자 수 가이드 (참고, 의미 우선):\n" + "\n".join(guide_lines)
|
|
|
|
# Phase Q-3: 글자수 예산 전달 (char_budget 우선, 없으면 Phase O 스펙)
|
|
char_budget = block.get("_char_budget", {})
|
|
container_h = block.get("_container_height_px")
|
|
|
|
if char_budget:
|
|
req_text += (
|
|
f"\n ★ 글자수 예산 (하드 제약 — 반드시 준수):"
|
|
f"\n - 최대 항목 수: {char_budget.get('max_items', '제한 없음')}개"
|
|
f"\n - 항목당 최대 글자 수: {char_budget.get('chars_per_item', '제한 없음')}자"
|
|
f"\n - 총 최대 글자 수: {char_budget.get('total_chars', '제한 없음')}자"
|
|
f"\n - 폰트 크기: {char_budget.get('font_size_px', 15.2)}px"
|
|
f"\n 이 예산은 컨테이너 크기에서 수학적으로 도출됨. 초과 시 overflow 발생."
|
|
)
|
|
elif container_h:
|
|
max_items = block.get("_max_items", "제한 없음")
|
|
max_chars_item = block.get("_max_chars_per_item", "제한 없음")
|
|
max_chars_total = block.get("_max_chars_total", "제한 없음")
|
|
font_size = block.get("_font_size_px", 15.2)
|
|
req_text += (
|
|
f"\n ★ 컨테이너 제약 (절대 준수):"
|
|
f"\n - 컨테이너 높이: {container_h}px"
|
|
f"\n - 최대 항목 수: {max_items}개"
|
|
f"\n - 항목당 최대 글자 수: {max_chars_item}자"
|
|
f"\n - 총 최대 글자 수: {max_chars_total}자"
|
|
f"\n - 폰트 크기: {font_size}px"
|
|
f"\n 이 제약을 넘기면 컨테이너 밖으로 넘친다. 반드시 지켜라."
|
|
)
|
|
|
|
slot_requirements.append(req_text)
|
|
|
|
page_label = ""
|
|
if len(layout_concept.get("pages", [])) > 1:
|
|
page_label = f" (페이지 {page_idx + 1}/{len(layout_concept['pages'])})"
|
|
|
|
# Phase M: 토픽별 source 정보 추출 (P-9 원본 보존 강화)
|
|
source_section = ""
|
|
if analysis:
|
|
source_lines = []
|
|
for topic in analysis.get("topics", []):
|
|
tid = topic.get("id")
|
|
hint = topic.get("source_hint", "")
|
|
data = topic.get("source_data", "")
|
|
if hint or data:
|
|
source_lines.append(
|
|
f"- 토픽 {tid} ({topic.get('purpose', '')}): "
|
|
f"{hint}{' / ' + data if data else ''}"
|
|
)
|
|
if source_lines:
|
|
source_section = (
|
|
"\n\n## 토픽별 원본 데이터 (이 텍스트에서 추출하라. 재작성 금지.)\n"
|
|
+ "\n".join(source_lines)
|
|
)
|
|
|
|
user_prompt = (
|
|
f"## 원본 콘텐츠 (참고용 — source_data가 있으면 source_data 우선)\n{content}\n\n"
|
|
f"## 블록 배치{page_label}\n"
|
|
+ "\n".join(slot_requirements)
|
|
+ source_section
|
|
+ "\n\n## 요청\n"
|
|
"각 블록의 ★★ source_data를 해당 블록의 슬롯에 그대로 배치하라.\n"
|
|
"source_data의 텍스트를 축약/요약/재작성하지 마라. 그대로 넣어라.\n"
|
|
"글자수 예산 초과 시에만 뒤에서부터 잘라내라.\n"
|
|
"형식:\n"
|
|
'{"blocks": [{"area": "...", "type": "...", "topic_id": 1, "data": {슬롯 키-값}}]}'
|
|
)
|
|
|
|
# Phase Q: 파싱 실패 시 재시도 (빈 data로 넘어가지 않는다)
|
|
import asyncio
|
|
MAX_FILL_RETRIES = 3
|
|
fill_success = False
|
|
|
|
for fill_attempt in range(MAX_FILL_RETRIES):
|
|
try:
|
|
result_text = await _call_kei_editor_with_retry(user_prompt)
|
|
|
|
filled = _parse_json(result_text)
|
|
|
|
if filled and "blocks" in filled:
|
|
filled_count = 0
|
|
for filled_block in filled["blocks"]:
|
|
matched = False
|
|
# 1차: topic_id로 정확 매칭
|
|
if filled_block.get("topic_id"):
|
|
for orig_block in blocks:
|
|
if orig_block.get("topic_id") == filled_block.get("topic_id"):
|
|
new_data = filled_block.get("data", {})
|
|
preserved = {}
|
|
if "data" in orig_block:
|
|
for k in ("column_override",):
|
|
if k in orig_block["data"]:
|
|
preserved[k] = orig_block["data"][k]
|
|
orig_block["data"] = {**new_data, **preserved}
|
|
matched = True
|
|
filled_count += 1
|
|
break
|
|
# 2차: area + type으로 매칭 (topic_id 없을 때)
|
|
if not matched:
|
|
for orig_block in blocks:
|
|
if (
|
|
orig_block.get("area") == filled_block.get("area")
|
|
and orig_block.get("type") == filled_block.get("type")
|
|
and "data" not in orig_block
|
|
):
|
|
new_data = filled_block.get("data", {})
|
|
preserved = {}
|
|
if "data" in orig_block:
|
|
for k in ("column_override",):
|
|
if k in orig_block["data"]:
|
|
preserved[k] = orig_block["data"][k]
|
|
orig_block["data"] = {**new_data, **preserved}
|
|
filled_count += 1
|
|
break
|
|
|
|
logger.info(
|
|
f"텍스트 정리 완료 (페이지 {page_idx + 1}): "
|
|
f"{filled_count}/{len(filled['blocks'])}개 블록 매칭"
|
|
)
|
|
|
|
# 검증: data가 실제로 채워진 블록이 있는가?
|
|
blocks_with_data = [b for b in blocks if b.get("data") and b.get("topic_id") is not None]
|
|
if blocks_with_data:
|
|
fill_success = True
|
|
break
|
|
else:
|
|
logger.warning(
|
|
f"[fill_content] 파싱 성공했으나 매칭된 블록 0개 "
|
|
f"(시도 {fill_attempt + 1}/{MAX_FILL_RETRIES})"
|
|
)
|
|
else:
|
|
logger.warning(
|
|
f"[fill_content] JSON 파싱 실패 (시도 {fill_attempt + 1}/{MAX_FILL_RETRIES}). "
|
|
f"응답: {result_text[:200] if result_text else '(비어있음)'}"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"텍스트 편집자 호출 실패 (시도 {fill_attempt + 1}): {e}")
|
|
if fill_attempt == MAX_FILL_RETRIES - 1:
|
|
raise
|
|
|
|
# 재시도 전 대기
|
|
if fill_attempt < MAX_FILL_RETRIES - 1:
|
|
await asyncio.sleep(5)
|
|
|
|
if not fill_success:
|
|
# 최대 재시도 후에도 실패 — 에러 발생 (빈 data로 진행하지 않음)
|
|
empty_blocks = [b.get("type") for b in blocks if not b.get("data") and b.get("topic_id") is not None]
|
|
raise RuntimeError(
|
|
f"fill_content 최대 재시도({MAX_FILL_RETRIES}회) 후에도 "
|
|
f"데이터 채우기 실패. 빈 블록: {empty_blocks}"
|
|
)
|
|
|
|
return layout_concept
|
|
|
|
|
|
async def _call_kei_editor_with_retry(prompt: str) -> str:
|
|
"""Kei API를 통해 텍스트 편집을 요청한다. 성공할 때까지 무한 재시도.
|
|
|
|
Kei persona의 도메인 지식 + RAG를 활용하여
|
|
건설/DX 분야 전문 용어를 정확하게 유지하면서 편집.
|
|
fallback 없음. Kei API가 응답할 때까지 기다린다.
|
|
"""
|
|
import asyncio
|
|
|
|
kei_url = getattr(settings, "kei_api_url", "http://localhost:8000")
|
|
full_prompt = EDITOR_PROMPT + "\n\n" + prompt
|
|
RETRY_INTERVAL = 10
|
|
attempt = 0
|
|
|
|
while True:
|
|
attempt += 1
|
|
try:
|
|
async with httpx.AsyncClient(timeout=None) as client:
|
|
async with client.stream(
|
|
"POST",
|
|
f"{kei_url}/api/direct",
|
|
json={
|
|
"message": full_prompt,
|
|
},
|
|
timeout=None,
|
|
) as response:
|
|
if response.status_code != 200:
|
|
logger.warning(f"Kei API (editor) HTTP {response.status_code} (시도 {attempt})")
|
|
await asyncio.sleep(RETRY_INTERVAL)
|
|
continue
|
|
|
|
full_text = await stream_sse_tokens(response)
|
|
|
|
if full_text:
|
|
return full_text
|
|
|
|
logger.warning(f"Kei API (editor) 텍스트 추출 실패 (시도 {attempt})")
|
|
await asyncio.sleep(RETRY_INTERVAL)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Kei API (editor) 호출 실패 (시도 {attempt}): {e}")
|
|
await asyncio.sleep(RETRY_INTERVAL)
|
|
|
|
|
|
|
|
async def fill_candidates(
|
|
content: str,
|
|
topic: dict[str, Any],
|
|
candidates: list[dict[str, Any]],
|
|
analysis: dict[str, Any] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Phase P: 1개 topic의 후보 3개 블록을 한꺼번에 텍스트 편집한다.
|
|
|
|
Kei 편집자 1회 호출로 3개 블록 각각의 슬롯에 맞게 편집.
|
|
|
|
Args:
|
|
content: 원본 텍스트
|
|
topic: 해당 topic 정보 (id, title, purpose, source_hint 등)
|
|
candidates: 후보 블록 3개 (type, _container_height_px, _max_items 등 포함)
|
|
analysis: 1단계 분석 결과
|
|
|
|
Returns:
|
|
candidates 리스트에 data가 채워진 상태로 반환
|
|
"""
|
|
tid = topic.get("id", "?")
|
|
purpose = topic.get("purpose", "")
|
|
source_hint = topic.get("source_hint", "")
|
|
source_data = topic.get("source_data", "")
|
|
|
|
# 각 후보 블록의 슬롯 + 컨테이너 스펙 정리
|
|
block_sections = []
|
|
for i, block in enumerate(candidates):
|
|
block_type = block.get("type", "")
|
|
slots = BLOCK_SLOTS.get(block_type, {})
|
|
|
|
section = (
|
|
f"### 후보 {i+1}: {block_type}\n"
|
|
f" 필수 슬롯: {slots.get('required', [])}\n"
|
|
f" 선택 슬롯: {slots.get('optional', [])}"
|
|
)
|
|
|
|
slot_desc = slots.get("slot_desc", {})
|
|
if slot_desc:
|
|
desc_lines = [f" {k}: {v}" for k, v in slot_desc.items()]
|
|
section += "\n 슬롯 설명:\n" + "\n".join(desc_lines)
|
|
|
|
# Phase R: expression_hint + variant 전달
|
|
if topic.get("expression_hint"):
|
|
section += f"\n ★ 표현 의도: {topic['expression_hint']}"
|
|
variant = block.get("_variant", "default")
|
|
if variant != "default":
|
|
section += f"\n ★ 변형: {variant}"
|
|
|
|
# Phase Q: 글자수 예산 전달 (있으면 우선, 없으면 Phase O 스펙)
|
|
char_budget = block.get("_char_budget", {})
|
|
container_h = block.get("_container_height_px")
|
|
|
|
if char_budget:
|
|
section += (
|
|
f"\n ★ 글자수 예산 (하드 제약 — 초과 시 overflow):"
|
|
f"\n 총 글자: {char_budget.get('total_chars', '제한 없음')}자"
|
|
f"\n 최대 항목: {char_budget.get('max_items', '제한 없음')}개"
|
|
f"\n 항목당 글자: {char_budget.get('chars_per_item', '제한 없음')}자"
|
|
)
|
|
elif container_h:
|
|
section += (
|
|
f"\n ★ 컨테이너 제약:"
|
|
f"\n 높이: {container_h}px"
|
|
f"\n 최대 항목: {block.get('_max_items', '제한 없음')}개"
|
|
f"\n 항목당 글자: {block.get('_max_chars_per_item', '제한 없음')}자"
|
|
f"\n 총 글자: {block.get('_max_chars_total', '제한 없음')}자"
|
|
)
|
|
|
|
block_sections.append(section)
|
|
|
|
source_section = ""
|
|
if source_hint or source_data:
|
|
source_section = (
|
|
f"\n\n## 원본 데이터 (이 텍스트에서 추출하라. 재작성 금지.)\n"
|
|
f" source_hint: {source_hint}\n"
|
|
f" source_data: {source_data}"
|
|
)
|
|
|
|
prompt = (
|
|
f"## 원본 콘텐츠\n{content}\n\n"
|
|
f"## 꼭지 {tid}: {topic.get('title', '')}\n"
|
|
f" 목적: {purpose}\n\n"
|
|
f"## 후보 블록 3개 — 각각의 슬롯에 맞게 텍스트를 편집하라\n\n"
|
|
+ "\n\n".join(block_sections)
|
|
+ source_section
|
|
+ "\n\n## 요청\n"
|
|
"위 3개 후보 블록 각각에 맞는 텍스트를 JSON으로 반환해줘.\n"
|
|
"원본에서 추출하라. 재작성 금지. 축약만 허용.\n"
|
|
"형식:\n"
|
|
'{"candidates": [\n'
|
|
' {"candidate_index": 0, "type": "블록타입", "data": {슬롯 키-값}},\n'
|
|
' {"candidate_index": 1, "type": "블록타입", "data": {슬롯 키-값}},\n'
|
|
' {"candidate_index": 2, "type": "블록타입", "data": {슬롯 키-값}}\n'
|
|
']}'
|
|
)
|
|
|
|
result_text = await _call_kei_editor_with_retry(prompt)
|
|
filled = _parse_json(result_text)
|
|
|
|
if filled and "candidates" in filled:
|
|
for filled_item in filled["candidates"]:
|
|
idx = filled_item.get("candidate_index", -1)
|
|
if 0 <= idx < len(candidates):
|
|
candidates[idx]["data"] = filled_item.get("data", {})
|
|
logger.info(f"[Phase P] 꼭지 {tid}: 후보 {len(filled['candidates'])}개 텍스트 편집 완료")
|
|
else:
|
|
logger.warning(f"[Phase P] 꼭지 {tid}: 텍스트 편집 파싱 실패")
|
|
|
|
return candidates
|
|
|
|
|
|
def _parse_json(text: str) -> dict[str, Any] | None:
|
|
"""텍스트에서 JSON을 추출한다.
|
|
|
|
Kei API가 마크다운 리스트 접두사(- )를 붙여 응답하는 경우에도 처리.
|
|
"""
|
|
# 전처리: 각 줄 앞의 마크다운 리스트 접두사(- ) 제거
|
|
lines = text.split("\n")
|
|
cleaned_lines = []
|
|
for line in lines:
|
|
stripped = line.lstrip()
|
|
if stripped.startswith("- "):
|
|
cleaned_lines.append(stripped[2:])
|
|
elif stripped.startswith("* "):
|
|
cleaned_lines.append(stripped[2:])
|
|
else:
|
|
cleaned_lines.append(stripped)
|
|
cleaned = "\n".join(cleaned_lines)
|
|
|
|
# 원본 먼저 시도 → 클린 버전 시도
|
|
for target in [text, cleaned]:
|
|
patterns = [
|
|
r"```json\s*(.*?)```",
|
|
r"```\s*(.*?)```",
|
|
r"(\{.*\})",
|
|
]
|
|
for pattern in patterns:
|
|
match = re.search(pattern, target, re.DOTALL)
|
|
if match:
|
|
try:
|
|
return json.loads(match.group(1).strip())
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return None
|