1021 lines
35 KiB
Python
1021 lines
35 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
step8_generate_report_gemini.py
|
|
|
|
기능
|
|
- 확정 목차(outline_issue_report.txt)를 읽어 섹션(소목차) 목록을 만든다.
|
|
- 섹션별로 RAG에서 근거 청크를 검색한다(FAISS 있으면 FAISS, 없으면 키워드 기반).
|
|
- 섹션별 본문 초안을 생성한다(내부 근거 우선, 원문 보존 원칙).
|
|
- 섹션별 이미지 후보를 매핑하고, md에는 이미지 자리표시자를 삽입한다.
|
|
- 산출물 2개를 만든다.
|
|
1) report_draft.md
|
|
2) report_sections.json
|
|
|
|
변경사항 (OpenAI → Gemini)
|
|
- google.genai 라이브러리 사용
|
|
- 자율성 통제: temperature=0.3, thinking_budget=0
|
|
- 원문 보존 원칙 강화
|
|
- 소목차별 중복 방지 로직 추가
|
|
- ★ 이미지 assets 복사 로직 추가
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
import shutil # ★ 추가: 이미지 복사용
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any, Optional, Tuple
|
|
|
|
import numpy as np
|
|
|
|
try:
|
|
import faiss # type: ignore
|
|
except Exception:
|
|
faiss = None
|
|
|
|
# ===== 하이브리드 API 설정 =====
|
|
# 검색/임베딩: OpenAI (기존 FAISS 인덱스 호환)
|
|
# 본문 작성: Gemini (글쓰기 품질)
|
|
|
|
from google import genai
|
|
from google.genai import types
|
|
from openai import OpenAI
|
|
from api_config import API_KEYS
|
|
|
|
# OpenAI (임베딩/검색용)
|
|
OPENAI_API_KEY = API_KEYS.get('GPT_API_KEY', '')
|
|
EMBED_MODEL = "text-embedding-3-small"
|
|
openai_client = OpenAI(api_key=OPENAI_API_KEY)
|
|
|
|
# Gemini (본문 작성용)
|
|
GEMINI_API_KEY = API_KEYS.get('GEMINI_API_KEY', '')
|
|
GEMINI_MODEL = "gemini-3-pro-preview"
|
|
gemini_client = genai.Client(api_key=GEMINI_API_KEY)
|
|
|
|
# ===== 경로 설정 =====
|
|
DATA_ROOT = Path(r"D:\for python\survey_test\process")
|
|
OUTPUT_ROOT = Path(r"D:\for python\survey_test\output")
|
|
CONTEXT_DIR = OUTPUT_ROOT / "context"
|
|
LOG_DIR = OUTPUT_ROOT / "logs"
|
|
RAG_DIR = OUTPUT_ROOT / "rag"
|
|
GEN_DIR = OUTPUT_ROOT / "generated"
|
|
|
|
# ★ 추가: 이미지 assets 경로
|
|
ASSETS_DIR = GEN_DIR / "assets"
|
|
IMAGES_ROOT = DATA_ROOT / "images" # 추출된 이미지 원본 위치
|
|
|
|
for d in [CONTEXT_DIR, LOG_DIR, RAG_DIR, GEN_DIR, ASSETS_DIR]:
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 파일명
|
|
OUTLINE_PATH = CONTEXT_DIR / "outline_issue_report.txt"
|
|
DOMAIN_PROMPT_PATH = CONTEXT_DIR / "domain_prompt.txt"
|
|
|
|
# 선택 파일(있으면 사용)
|
|
FAISS_INDEX_PATH = RAG_DIR / "faiss.index"
|
|
FAISS_META_PATH = RAG_DIR / "meta.json"
|
|
FAISS_VECTORS_PATH = RAG_DIR / "vectors.npy"
|
|
|
|
# 이미지 메타(있으면 캡션 보강)
|
|
IMAGE_META_PATH = DATA_ROOT / "image_metadata.json"
|
|
|
|
# 출력 파일
|
|
REPORT_MD_PATH = GEN_DIR / "report_draft.md"
|
|
REPORT_JSON_PATH = GEN_DIR / "report_sections.json"
|
|
|
|
# 설정값
|
|
TOP_K_EVIDENCE = int(os.getenv("TOP_K_EVIDENCE", "10"))
|
|
MAX_IMAGES_PER_SECTION = int(os.getenv("MAX_IMAGES_PER_SECTION", "3"))
|
|
MAX_EVIDENCE_SNIPPET_CHARS = int(os.getenv("MAX_EVIDENCE_SNIPPET_CHARS", "900"))
|
|
|
|
# 패턴
|
|
RE_TITLE_LINE = re.compile(r"^\s*(.+?)\s*$")
|
|
RE_L1 = re.compile(r"^\s*(\d+)\.\s+(.+?)\s*$")
|
|
RE_L2 = re.compile(r"^\s*(\d+\.\d+)\s+(.+?)\s*$")
|
|
RE_L3 = re.compile(r"^\s*(\d+\.\d+\.\d+)\s+(.+?)\s*$")
|
|
RE_KEYWORDS = re.compile(r"(#\S+)")
|
|
|
|
RE_IMAGE_PATH_IN_MD = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
|
|
|
|
|
|
def log(msg: str):
|
|
line = f"[{datetime.now().strftime('%H:%M:%S')}] {msg}"
|
|
print(line, flush=True)
|
|
with (LOG_DIR / "step8_generate_report_log.txt").open("a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
|
|
|
|
@dataclass
|
|
class SubTopic:
|
|
title: str
|
|
keywords: List[str]
|
|
type: str
|
|
guide: str
|
|
|
|
|
|
@dataclass
|
|
class OutlineItem:
|
|
number: str
|
|
title: str
|
|
depth: int
|
|
sub_topics: List[SubTopic] = field(default_factory=list)
|
|
|
|
|
|
def read_text(p: Path) -> str:
|
|
return p.read_text(encoding="utf-8", errors="ignore").strip()
|
|
|
|
|
|
def load_domain_prompt() -> str:
|
|
if not DOMAIN_PROMPT_PATH.exists():
|
|
raise RuntimeError(f"domain_prompt.txt 없음: {DOMAIN_PROMPT_PATH}")
|
|
return read_text(DOMAIN_PROMPT_PATH)
|
|
|
|
|
|
def load_outline() -> Tuple[str, List[OutlineItem]]:
|
|
if not OUTLINE_PATH.exists():
|
|
raise RuntimeError("목차 파일이 없습니다.")
|
|
raw = OUTLINE_PATH.read_text(encoding="utf-8", errors="ignore").splitlines()
|
|
if not raw:
|
|
return "", []
|
|
|
|
report_title = raw[0].strip()
|
|
items: List[OutlineItem] = []
|
|
current_l3 = None
|
|
|
|
# 꼭지 파싱용 정규식
|
|
re_l3_head = re.compile(r"^\s*(\d+\.\d+\.\d+)\s+(.+)$")
|
|
re_l3_topic = re.compile(r"^\s*[\-\*]\s+(.+?)\s*\|\s*(.+?)\s*\|\s*(\[.+?\])\s*\|\s*(.+)$")
|
|
|
|
for ln in raw[1:]:
|
|
line = ln.strip()
|
|
if not line:
|
|
continue
|
|
|
|
m3h = re_l3_head.match(line)
|
|
if m3h:
|
|
current_l3 = OutlineItem(number=m3h.group(1), title=m3h.group(2), depth=3)
|
|
items.append(current_l3)
|
|
continue
|
|
|
|
m3t = re_l3_topic.match(line)
|
|
if m3t and current_l3:
|
|
kws = [k.lstrip("#").strip() for k in RE_KEYWORDS.findall(m3t.group(2))]
|
|
current_l3.sub_topics.append(SubTopic(
|
|
title=m3t.group(1), keywords=kws, type=m3t.group(3), guide=m3t.group(4)
|
|
))
|
|
continue
|
|
|
|
m2 = RE_L2.match(line)
|
|
if m2:
|
|
items.append(OutlineItem(number=m2.group(1), title=m2.group(2), depth=2))
|
|
current_l3 = None
|
|
continue
|
|
m1 = RE_L1.match(line)
|
|
if m1:
|
|
items.append(OutlineItem(number=m1.group(1), title=m1.group(2), depth=1))
|
|
current_l3 = None
|
|
continue
|
|
|
|
return report_title, items
|
|
|
|
|
|
def load_image_metadata() -> Dict[str, Dict[str, Any]]:
|
|
"""image_metadata.json이 있으면 image_file 기준으로 맵을 만든다."""
|
|
if not IMAGE_META_PATH.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(IMAGE_META_PATH.read_text(encoding="utf-8", errors="ignore"))
|
|
out: Dict[str, Dict[str, Any]] = {}
|
|
for it in data:
|
|
fn = (it.get("image_file") or "").strip()
|
|
if fn:
|
|
out[fn] = it
|
|
return out
|
|
except Exception as e:
|
|
log(f"[WARN] image_metadata.json 로드 실패: {e}")
|
|
return {}
|
|
|
|
|
|
def iter_rag_items() -> List[Dict[str, Any]]:
|
|
"""rag 폴더의 *_chunks.json 모두 로드"""
|
|
items: List[Dict[str, Any]] = []
|
|
files = sorted(RAG_DIR.glob("*_chunks.json"))
|
|
if not files:
|
|
raise RuntimeError(f"rag 폴더에 *_chunks.json 없음: {RAG_DIR}")
|
|
|
|
for f in files:
|
|
try:
|
|
data = json.loads(f.read_text(encoding="utf-8", errors="ignore"))
|
|
if isinstance(data, list):
|
|
for it in data:
|
|
if isinstance(it, dict):
|
|
items.append(it)
|
|
except Exception as e:
|
|
log(f"[WARN] RAG 파일 로드 실패: {f.name} {e}")
|
|
|
|
return items
|
|
|
|
|
|
def normalize_ws(s: str) -> str:
|
|
return " ".join((s or "").split())
|
|
|
|
|
|
def make_evidence_snippet(text: str, max_chars: int) -> str:
|
|
t = normalize_ws(text)
|
|
if len(t) <= max_chars:
|
|
return t
|
|
return t[:max_chars] + "..."
|
|
|
|
|
|
def get_item_key(it: Dict[str, Any]) -> Tuple[str, int]:
|
|
src = (it.get("source") or "").strip()
|
|
ch = int(it.get("chunk") or 0)
|
|
return (src, ch)
|
|
|
|
|
|
def build_item_index(items: List[Dict[str, Any]]) -> Dict[Tuple[str, int], Dict[str, Any]]:
|
|
m: Dict[Tuple[str, int], Dict[str, Any]] = {}
|
|
for it in items:
|
|
m[get_item_key(it)] = it
|
|
return m
|
|
|
|
|
|
def try_load_faiss():
|
|
"""faiss.index, meta.json, vectors.npy가 모두 있고 faiss 모듈이 있으면 사용"""
|
|
if faiss is None:
|
|
log("[INFO] faiss 모듈 없음 - 키워드 검색 사용")
|
|
return None
|
|
if not (FAISS_INDEX_PATH.exists() and FAISS_META_PATH.exists() and FAISS_VECTORS_PATH.exists()):
|
|
log("[INFO] FAISS 파일 없음 - 키워드 검색 사용")
|
|
return None
|
|
try:
|
|
index = faiss.read_index(str(FAISS_INDEX_PATH))
|
|
metas = json.loads(FAISS_META_PATH.read_text(encoding="utf-8", errors="ignore"))
|
|
vecs = np.load(str(FAISS_VECTORS_PATH))
|
|
log(f"[INFO] FAISS 로드 성공 - 인덱스 차원: {index.d}, 메타 수: {len(metas)}")
|
|
return index, metas, vecs
|
|
except Exception as e:
|
|
log(f"[WARN] FAISS 로드 실패: {e}")
|
|
return None
|
|
|
|
|
|
def embed_query_openai(q: str) -> np.ndarray:
|
|
"""OpenAI 임베딩 (기존 FAISS 인덱스와 호환)"""
|
|
try:
|
|
resp = openai_client.embeddings.create(model=EMBED_MODEL, input=[q])
|
|
v = np.array(resp.data[0].embedding, dtype="float32")
|
|
n = np.linalg.norm(v) + 1e-12
|
|
return v / n
|
|
except Exception as e:
|
|
log(f"[WARN] OpenAI 임베딩 실패: {e}")
|
|
return np.zeros(1536, dtype="float32") # OpenAI 차원
|
|
|
|
|
|
def retrieve_with_faiss(
|
|
index,
|
|
metas: List[Dict[str, Any]],
|
|
item_map: Dict[Tuple[str, int], Dict[str, Any]],
|
|
query: str,
|
|
top_k: int
|
|
) -> List[Dict[str, Any]]:
|
|
qv = embed_query_openai(query).reshape(1, -1).astype("float32")
|
|
D, I = index.search(qv, top_k)
|
|
out: List[Dict[str, Any]] = []
|
|
for idx in I[0]:
|
|
if idx < 0 or idx >= len(metas):
|
|
continue
|
|
meta = metas[idx]
|
|
src = (meta.get("source") or "").strip()
|
|
ch = int(meta.get("chunk") or 0)
|
|
it = item_map.get((src, ch))
|
|
if it:
|
|
out.append(it)
|
|
return out
|
|
|
|
|
|
def tokenize_simple(s: str) -> List[str]:
|
|
s = normalize_ws(s).lower()
|
|
return [t for t in re.split(r"\s+", s) if t]
|
|
|
|
|
|
def retrieve_with_keywords(
|
|
all_items: List[Dict[str, Any]],
|
|
query: str,
|
|
keywords: List[str],
|
|
top_k: int
|
|
) -> List[Dict[str, Any]]:
|
|
q_tokens = set(tokenize_simple(query))
|
|
k_tokens = set([kw.lower() for kw in keywords if kw])
|
|
|
|
scored: List[Tuple[float, Dict[str, Any]]] = []
|
|
for it in all_items:
|
|
txt = " ".join([
|
|
str(it.get("title") or ""),
|
|
str(it.get("keywords") or ""),
|
|
str(it.get("summary") or ""),
|
|
str(it.get("text") or ""),
|
|
str(it.get("folder_context") or ""),
|
|
str(it.get("source_path") or ""),
|
|
])
|
|
t = normalize_ws(txt).lower()
|
|
|
|
score = 0.0
|
|
for tok in q_tokens:
|
|
if tok and tok in t:
|
|
score += 1.0
|
|
for tok in k_tokens:
|
|
if tok and tok in t:
|
|
score += 2.0
|
|
|
|
if it.get("has_images"):
|
|
score += 0.5
|
|
|
|
if score > 0:
|
|
scored.append((score, it))
|
|
|
|
scored.sort(key=lambda x: x[0], reverse=True)
|
|
return [it for _, it in scored[:top_k]]
|
|
|
|
|
|
def select_images_for_section(
|
|
evidences: List[Dict[str, Any]],
|
|
image_meta_by_file: Dict[str, Dict[str, Any]],
|
|
max_images: int
|
|
) -> List[Dict[str, Any]]:
|
|
"""근거 청크에서 images를 모아 섹션 이미지 후보를 만들고 상한으로 자른다."""
|
|
seen = set()
|
|
out: List[Dict[str, Any]] = []
|
|
|
|
def infer_image_file(p: str) -> str:
|
|
p = p.replace("\\", "/")
|
|
return p.split("/")[-1]
|
|
|
|
for ev in evidences:
|
|
imgs = ev.get("images") or []
|
|
if not isinstance(imgs, list):
|
|
continue
|
|
for img in imgs:
|
|
if not isinstance(img, dict):
|
|
continue
|
|
rel_path = (img.get("path") or "").strip()
|
|
if not rel_path:
|
|
continue
|
|
key = rel_path.replace("\\", "/")
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
|
|
img_file = infer_image_file(key)
|
|
meta = image_meta_by_file.get(img_file, {})
|
|
|
|
caption = ""
|
|
if meta:
|
|
caption = (meta.get("caption") or "").strip()
|
|
if not caption:
|
|
caption = (img.get("alt") or "").strip() or img_file
|
|
|
|
out.append({
|
|
"image_id": "",
|
|
"rel_path": key,
|
|
"image_file": img_file,
|
|
"caption": caption,
|
|
"source_path": ev.get("source_path") or ev.get("source") or "",
|
|
"page": meta.get("page", None) if meta else None,
|
|
"type": meta.get("type", None) if meta else None,
|
|
})
|
|
if len(out) >= max_images:
|
|
return out
|
|
|
|
return out
|
|
|
|
|
|
def make_image_placeholders(section_number: str, images: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""섹션번호 기반으로 이미지아이디를 만들고 placeholder를 만든다."""
|
|
sec_key = section_number.replace(".", "_")
|
|
out = []
|
|
for i, img in enumerate(images, start=1):
|
|
img_id = f"{sec_key}_img{i:02d}"
|
|
out.append({**img, "image_id": img_id, "placeholder": f"{{{{IMG:{img_id}}}}}"})
|
|
return out
|
|
|
|
|
|
# ★ 추가: 이미지 파일을 assets 폴더로 복사하는 함수
|
|
def copy_images_to_assets(image_info_list: List[Dict[str, Any]]) -> None:
|
|
"""선택된 이미지들을 generated/assets/로 복사"""
|
|
for img in image_info_list:
|
|
# 원본 경로 찾기 (여러 경로 시도)
|
|
rel_path = img.get('rel_path', '')
|
|
src_path = None
|
|
|
|
# 1차: DATA_ROOT 기준 상대경로
|
|
candidate1 = DATA_ROOT / rel_path
|
|
if candidate1.exists():
|
|
src_path = candidate1
|
|
|
|
# 2차: IMAGES_ROOT에서 파일명으로 검색
|
|
if src_path is None:
|
|
candidate2 = IMAGES_ROOT / img.get('image_file', '')
|
|
if candidate2.exists():
|
|
src_path = candidate2
|
|
|
|
# 3차: DATA_ROOT 전체에서 파일명 검색 (재귀)
|
|
if src_path is None:
|
|
img_file = img.get('image_file', '')
|
|
if img_file:
|
|
for found in DATA_ROOT.rglob(img_file):
|
|
src_path = found
|
|
break
|
|
|
|
if src_path and src_path.exists():
|
|
# image_id 기반으로 새 파일명 생성 (확장자 유지)
|
|
ext = src_path.suffix or '.png'
|
|
dst_filename = f"{img['image_id']}{ext}"
|
|
dst_path = ASSETS_DIR / dst_filename
|
|
|
|
try:
|
|
shutil.copy2(src_path, dst_path)
|
|
img['asset_path'] = f"assets/{dst_filename}"
|
|
log(f" [IMG] {img['image_id']} → {dst_filename}")
|
|
except Exception as e:
|
|
log(f" [WARN] 이미지 복사 실패: {img['image_id']} - {e}")
|
|
img['asset_path'] = None
|
|
else:
|
|
log(f" [WARN] 이미지 없음: {rel_path} ({img.get('image_file', '')})")
|
|
img['asset_path'] = None
|
|
|
|
|
|
# ===== Gemini 프롬프트 구성 (자율성 통제 강화) =====
|
|
|
|
def build_system_instruction(domain_prompt: str) -> str:
|
|
"""
|
|
Gemini 시스템 지시문 (v4 - 최종)
|
|
"""
|
|
return f"""{domain_prompt}
|
|
|
|
═══════════════════════════════════════════════════════════════
|
|
★★★ 절대 준수 규칙 ★★★
|
|
═══════════════════════════════════════════════════════════════
|
|
|
|
[금지 사항]
|
|
1. 원문의 수치, 용어, 표현을 임의로 변경 금지
|
|
2. 제공되지 않은 정보 추론/창작 금지
|
|
3. 추측성 표현 금지 ("~로 보인다", "~일 것이다")
|
|
4. 중복 내용 작성 금지
|
|
5. 마크다운 헤딩(#, ##, ###, ####) 사용 금지
|
|
6. ★ "꼭지", "항목 1", "Topic" 등 내부 분류 용어 출력 금지
|
|
7. ★ "1. 2. 3." 형태 번호 사용 금지 (반드시 "1) 2) 3)" 사용)
|
|
|
|
[필수 사항]
|
|
1. 원문 최대 보존
|
|
2. 수치는 원본 그대로
|
|
3. 전문 용어 변경 없이 사용
|
|
4. 보고서 형식으로 전문적 작성
|
|
|
|
═══════════════════════════════════════════════════════════════
|
|
★★★ 번호 체계 및 서식 규칙 (필수) ★★★
|
|
═══════════════════════════════════════════════════════════════
|
|
|
|
【레벨별 번호와 서식】
|
|
|
|
■ 1단계: 1), 2), 3)
|
|
■ 2단계: (1), (2), (3)
|
|
■ 3단계: ①, ②, ③ 또는 -, *
|
|
|
|
【핵심 서식 규칙】
|
|
|
|
★ 모든 번호의 제목은 반드시 **볼드** 처리
|
|
★ 제목과 본문 사이에 반드시 빈 줄(엔터) 삽입
|
|
★ 본문과 다음 번호 사이에 반드시 빈 줄(엔터) 삽입
|
|
|
|
【올바른 예시】
|
|
```
|
|
1) **VRS GNSS 측량의 개요**
|
|
|
|
인공위성과 위성기준점을 이용한 위치 측량 방식이다. 실시간 보정을 통해 높은 정확도를 확보할 수 있다.
|
|
|
|
2) **UAV 사진측량의 특징**
|
|
|
|
무인항공기를 활용한 광역 측량 방식이다. 목적에 따라 다음과 같이 구분된다.
|
|
|
|
(1) **맵핑측량**
|
|
|
|
정사영상 제작에 특화된 촬영 방식이다.
|
|
|
|
(2) **모델측량**
|
|
|
|
3D 모델 생성에 특화된 촬영 방식이다.
|
|
```
|
|
|
|
【잘못된 예시 - 절대 금지】
|
|
```
|
|
꼭지 1 VRS GNSS 측량 ← "꼭지" 용어 금지!
|
|
1. VRS GNSS 측량 ← "1." 형태 금지!
|
|
1) VRS GNSS 측량 인공위성을... ← 제목+본문 한줄 금지!
|
|
1) VRS GNSS 측량 ← 볼드 없음 금지!
|
|
```
|
|
|
|
═══════════════════════════════════════════════════════════════
|
|
|
|
[작성 형식]
|
|
- 섹션 제목 없이 바로 본문 시작
|
|
- 주제별 구분: 1), 2), 3) + **볼드 제목** + 줄바꿈 + 본문
|
|
- 하위 구분: (1), (2), (3) + **볼드 제목** + 줄바꿈 + 본문
|
|
- [비교형]: 마크다운 표 포함
|
|
- [기술형]: 기술 사양/수치 정확히 기재
|
|
- [절차형]: 단계별 1), 2), 3) 사용
|
|
|
|
[출력 제한]
|
|
- 마크다운 헤딩 금지
|
|
- "꼭지", "Topic", "항목" 등 분류 용어 출력 금지
|
|
- 내부 메모용 표현 금지
|
|
- 출처 표시 금지
|
|
═══════════════════════════════════════════════════════════════
|
|
"""
|
|
|
|
|
|
def build_user_prompt(
|
|
report_title: str,
|
|
item, # OutlineItem
|
|
evidences,
|
|
image_info_list,
|
|
previous_sections_summary: str = ""
|
|
) -> str:
|
|
"""
|
|
섹션별 사용자 프롬프트 (v4)
|
|
"""
|
|
|
|
# 근거 자료 정리
|
|
ev_text = ""
|
|
for i, ev in enumerate(evidences, 1):
|
|
src = ev.get('source_path') or ev.get('source', '내부자료')
|
|
text = ev.get('text', '')[:1500]
|
|
title = ev.get('title', '')
|
|
keywords = ev.get('keywords', '')
|
|
|
|
ev_text += f"""
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
[데이터 {i}] 출처: {src}
|
|
제목: {title}
|
|
키워드: {keywords}
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
{text}
|
|
"""
|
|
|
|
# ★ "꼭지" → "주제"로 변경, 번호 부여
|
|
topic_guides = ""
|
|
for idx, st in enumerate(item.sub_topics, 1):
|
|
topic_guides += f"""
|
|
【작성할 내용 {idx}】 {st.title}
|
|
- 유형: {st.type}
|
|
- 핵심 키워드: {', '.join(['#'+k for k in st.keywords]) if st.keywords else '없음'}
|
|
- 참고 지침: {st.guide}
|
|
- ★ 출력 시 "{idx}) **{st.title}**" 형태로 시작할 것
|
|
"""
|
|
|
|
# 이미지 안내
|
|
img_guide = ""
|
|
if image_info_list:
|
|
img_guide = "\n【삽입 가능 이미지】\n"
|
|
for img in image_info_list:
|
|
img_guide += f" - {img['placeholder']}: {img['caption']}\n"
|
|
img_guide += " → 문맥에 맞는 위치에 삽입\n"
|
|
|
|
# 중복 방지
|
|
dup_guide = ""
|
|
if previous_sections_summary:
|
|
dup_guide = f"""
|
|
【중복 방지 - 이미 다룬 내용이므로 제외】
|
|
{previous_sections_summary}
|
|
"""
|
|
|
|
# ★ 서식 리마인더 강화
|
|
format_reminder = """
|
|
═══════════════════════════════════════════════════════════════
|
|
★★★ 출력 서식 필수 준수 ★★★
|
|
═══════════════════════════════════════════════════════════════
|
|
1) **제목은 반드시 볼드**
|
|
|
|
본문은 제목 다음 줄에 작성
|
|
|
|
2) **다음 제목도 볼드**
|
|
|
|
본문...
|
|
|
|
(1) **하위 제목도 볼드**
|
|
|
|
하위 본문...
|
|
|
|
★ "꼭지", "항목", "Topic" 등 내부 용어 절대 출력 금지!
|
|
★ 제목과 본문 사이 반드시 빈 줄!
|
|
═══════════════════════════════════════════════════════════════
|
|
"""
|
|
|
|
return f"""
|
|
╔═══════════════════════════════════════════════════════════════╗
|
|
║ 보고서: {report_title}
|
|
║ 작성 섹션: {item.number} {item.title}
|
|
╚═══════════════════════════════════════════════════════════════╝
|
|
|
|
{dup_guide}
|
|
|
|
【이 섹션에서 다룰 내용】
|
|
{topic_guides}
|
|
|
|
{img_guide}
|
|
|
|
{format_reminder}
|
|
|
|
【참고 데이터】
|
|
{ev_text}
|
|
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
지시: '{item.number} {item.title}' 섹션 본문을 작성하라.
|
|
|
|
★ 번호: 1), 2) → (1), (2) → -, *
|
|
★ 제목: 반드시 **볼드**
|
|
★ 줄바꿈: 제목↔본문 사이 빈 줄 필수
|
|
★ 금지어: "꼭지", "항목", "Topic" 출력 금지
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
"""
|
|
|
|
|
|
def generate_section_text_gemini(
|
|
system_instruction: str,
|
|
user_prompt: str
|
|
) -> str:
|
|
"""
|
|
Gemini API를 사용한 섹션 본문 생성
|
|
- temperature=0.3으로 자율성 억제
|
|
"""
|
|
try:
|
|
response = gemini_client.models.generate_content(
|
|
model=GEMINI_MODEL,
|
|
contents=user_prompt,
|
|
config=types.GenerateContentConfig(
|
|
system_instruction=system_instruction,
|
|
temperature=0.3, # 낮은 temperature로 창의성 억제
|
|
)
|
|
)
|
|
return (response.text or "").strip()
|
|
except Exception as e:
|
|
log(f"[ERROR] Gemini API 호출 실패: {e}")
|
|
return f"[생성 실패: {e}]"
|
|
|
|
import re
|
|
|
|
def extract_section_summary(text: str, max_chars: int = 200) -> str:
|
|
"""섹션 본문에서 핵심 키워드/주제 추출 (중복 방지용)"""
|
|
# 첫 200자 또는 첫 문단
|
|
lines = text.split('\n')
|
|
summary_parts = []
|
|
char_count = 0
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
summary_parts.append(line)
|
|
char_count += len(line)
|
|
if char_count >= max_chars:
|
|
break
|
|
|
|
return ' '.join(summary_parts)[:max_chars]
|
|
|
|
|
|
def fix_numbering_format(text: str) -> str:
|
|
"""
|
|
Gemini가 "1. 2. 3." 형태로 출력했을 때 "1) 2) 3)" 형태로 변환
|
|
|
|
변환 규칙:
|
|
- "1. " → "1) " (줄 시작, 들여쓰기 0)
|
|
- " 1. " → " (1) " (들여쓰기 있으면 하위 레벨)
|
|
"""
|
|
lines = text.split('\n')
|
|
result = []
|
|
|
|
for line in lines:
|
|
# 원본 들여쓰기 측정
|
|
stripped = line.lstrip()
|
|
indent = len(line) - len(stripped)
|
|
|
|
# "숫자. " 패턴 감지 (마크다운 순서 리스트)
|
|
match = re.match(r'^(\d+)\.\s+(.+)$', stripped)
|
|
|
|
if match:
|
|
num = match.group(1)
|
|
content = match.group(2)
|
|
|
|
if indent == 0:
|
|
# 최상위 레벨: 1. → 1)
|
|
result.append(f"{num}) {content}")
|
|
elif indent <= 4:
|
|
# 1단계 들여쓰기: 1. → (1)
|
|
result.append(" " * indent + f"({num}) {content}")
|
|
else:
|
|
# 2단계 이상 들여쓰기: 그대로 유지 또는 - 로 변환
|
|
result.append(" " * indent + f"- {content}")
|
|
else:
|
|
result.append(line)
|
|
|
|
return '\n'.join(result)
|
|
|
|
|
|
def clean_generated_text_final(section_number: str, text: str) -> str:
|
|
"""
|
|
Gemini 출력 후처리 (최종 버전)
|
|
|
|
1. 중복 섹션 제목 제거
|
|
2. "꼭지 N" 패턴 제거
|
|
3. 번호 체계 변환 (1. → 1))
|
|
4. 제목 볼드 + 줄바꿈 강제 적용
|
|
5. #### 헤딩 → 볼드 변환
|
|
"""
|
|
|
|
# 1단계: 기본 정리
|
|
lines = text.split('\n')
|
|
cleaned = []
|
|
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
|
|
# 중복 섹션 제목 제거 (# 숫자.숫자.숫자 형태)
|
|
if re.match(r'^#{1,4}\s*\d+(\.\d+)*\s+', stripped):
|
|
continue
|
|
|
|
# "꼭지 N" 패턴 제거 (독립 라인)
|
|
if re.match(r'^[\*\*]*꼭지\s*\d+[\*\*]*\s*', stripped):
|
|
continue
|
|
|
|
# "**꼭지 N 제목**" → "**제목**" 변환
|
|
cleaned_line = re.sub(r'\*\*꼭지\s*\d+\s*', '**', stripped)
|
|
|
|
# #### 헤딩 → 볼드
|
|
h4_match = re.match(r'^####\s+(.+)$', cleaned_line)
|
|
if h4_match:
|
|
title = h4_match.group(1).strip()
|
|
if not re.match(r'^\d+', title):
|
|
cleaned.append(f"\n**{title}**\n")
|
|
continue
|
|
|
|
# 빈 줄 연속 방지 (3줄 이상 → 2줄)
|
|
if not stripped:
|
|
if len(cleaned) >= 2 and not cleaned[-1].strip() and not cleaned[-2].strip():
|
|
continue
|
|
|
|
cleaned.append(cleaned_line if cleaned_line != stripped else line)
|
|
|
|
result = '\n'.join(cleaned)
|
|
|
|
# 2단계: 번호 체계 변환
|
|
result = fix_numbering_format(result)
|
|
|
|
# 3단계: 제목+본문 붙어있는 것 분리 + 볼드 적용
|
|
result = fix_title_format(result)
|
|
|
|
return result.strip()
|
|
|
|
|
|
def fix_numbering_format(text: str) -> str:
|
|
"""
|
|
"1. " → "1) " 변환
|
|
들여쓰기 있으면 "(1)" 형태로
|
|
"""
|
|
lines = text.split('\n')
|
|
result = []
|
|
|
|
for line in lines:
|
|
stripped = line.lstrip()
|
|
indent = len(line) - len(stripped)
|
|
|
|
# "숫자. " 패턴 (마크다운 순서 리스트)
|
|
match = re.match(r'^(\d+)\.\s+(.+)$', stripped)
|
|
|
|
if match:
|
|
num = match.group(1)
|
|
content = match.group(2)
|
|
|
|
if indent == 0:
|
|
# 최상위: 1. → 1)
|
|
result.append(f"{num}) {content}")
|
|
elif indent <= 4:
|
|
# 1단계 들여쓰기: → (1)
|
|
result.append(" " * indent + f"({num}) {content}")
|
|
else:
|
|
# 2단계 이상: → -
|
|
result.append(" " * indent + f"- {content}")
|
|
else:
|
|
result.append(line)
|
|
|
|
return '\n'.join(result)
|
|
|
|
|
|
def fix_title_format(text: str) -> str:
|
|
"""
|
|
번호+제목+본문 한줄 → 번호+제목 / 본문 분리
|
|
제목에 볼드 적용
|
|
|
|
핵심: **볼드 제목** 뒤에 본문이 이어지면 줄바꿈 삽입
|
|
"""
|
|
lines = text.split('\n')
|
|
result = []
|
|
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
indent = len(line) - len(stripped)
|
|
indent_str = " " * indent
|
|
|
|
# 패턴 1: "1) **제목** 본문..." → "1) **제목**\n\n본문..."
|
|
m1 = re.match(r'^(\d+)\)\s+(\*\*[^*]+\*\*)\s+(.{20,})$', stripped)
|
|
if m1:
|
|
num = m1.group(1)
|
|
title = m1.group(2)
|
|
body = m1.group(3).strip()
|
|
result.append(f"{indent_str}{num}) {title}")
|
|
result.append("")
|
|
result.append(f"{indent_str}{body}")
|
|
result.append("")
|
|
continue
|
|
|
|
# 패턴 2: "(1) **제목** 본문..." → "(1) **제목**\n\n본문..."
|
|
m2 = re.match(r'^\((\d+)\)\s+(\*\*[^*]+\*\*)\s+(.{20,})$', stripped)
|
|
if m2:
|
|
num = m2.group(1)
|
|
title = m2.group(2)
|
|
body = m2.group(3).strip()
|
|
result.append(f"{indent_str}({num}) {title}")
|
|
result.append("")
|
|
result.append(f"{indent_str}{body}")
|
|
result.append("")
|
|
continue
|
|
|
|
# 패턴 3: "1) 제목:" 또는 "1) 제목" (볼드 없음, 짧은 제목) → 볼드 적용
|
|
m3 = re.match(r'^(\d+)\)\s+([^*\n]{3,40})$', stripped)
|
|
if m3:
|
|
num = m3.group(1)
|
|
title = m3.group(2).strip().rstrip(':')
|
|
# 문장이 아닌 제목으로 판단 (마침표로 안 끝남)
|
|
if not title.endswith(('.', '다', '요', '음', '함')):
|
|
result.append(f"{indent_str}{num}) **{title}**")
|
|
result.append("")
|
|
continue
|
|
|
|
# 패턴 4: "(1) 제목" (볼드 없음) → 볼드 적용
|
|
m4 = re.match(r'^\((\d+)\)\s+([^*\n]{3,40})$', stripped)
|
|
if m4:
|
|
num = m4.group(1)
|
|
title = m4.group(2).strip().rstrip(':')
|
|
if not title.endswith(('.', '다', '요', '음', '함')):
|
|
result.append(f"{indent_str}({num}) **{title}**")
|
|
result.append("")
|
|
continue
|
|
|
|
result.append(line)
|
|
|
|
# 연속 빈줄 정리
|
|
final = []
|
|
for line in result:
|
|
if not line.strip():
|
|
if len(final) >= 2 and not final[-1].strip() and not final[-2].strip():
|
|
continue
|
|
final.append(line)
|
|
|
|
return '\n'.join(final)
|
|
|
|
|
|
def main():
|
|
log("=== step8 Gemini 기반 보고서 생성 시작 ===")
|
|
|
|
domain_prompt = load_domain_prompt()
|
|
report_title, outline_items = load_outline()
|
|
|
|
log(f"보고서 제목: {report_title}")
|
|
log(f"목차 항목 수: {len(outline_items)}")
|
|
|
|
# 데이터 및 이미지 메타 로드
|
|
image_meta_by_file = load_image_metadata()
|
|
all_rag_items = iter_rag_items()
|
|
item_map = build_item_index(all_rag_items)
|
|
faiss_pack = try_load_faiss()
|
|
use_faiss = faiss_pack is not None
|
|
|
|
log(f"RAG 청크 수: {len(all_rag_items)}")
|
|
log(f"FAISS 사용: {use_faiss}")
|
|
|
|
# 시스템 지시문 (한 번만 생성)
|
|
system_instruction = build_system_instruction(domain_prompt)
|
|
|
|
md_lines = [f"# {report_title}", ""]
|
|
report_json_sections = []
|
|
|
|
# 중복 방지를 위한 이전 섹션 요약 누적
|
|
previous_sections_summary = ""
|
|
|
|
# ★ 추가: 복사된 이미지 카운트
|
|
total_images_copied = 0
|
|
|
|
for it in outline_items:
|
|
# 대목차와 중목차는 제목만 적고 통과
|
|
if it.depth < 3:
|
|
prefix = "## " if it.depth == 1 else "### "
|
|
md_lines.append(f"\n{prefix}{it.number} {it.title}\n")
|
|
continue
|
|
|
|
log(f"집필 중: {it.number} {it.title} (꼭지 {len(it.sub_topics)}개)")
|
|
|
|
# 꼭지들의 키워드를 합쳐서 검색
|
|
all_kws = []
|
|
for st in it.sub_topics:
|
|
all_kws.extend(st.keywords)
|
|
query = f"{it.title} " + " ".join(all_kws)
|
|
|
|
# RAG 검색
|
|
if use_faiss:
|
|
evidences = retrieve_with_faiss(faiss_pack[0], faiss_pack[1], item_map, query, 12)
|
|
else:
|
|
evidences = retrieve_with_keywords(all_rag_items, query, all_kws, 12)
|
|
|
|
log(f" → 검색된 근거 청크: {len(evidences)}개")
|
|
|
|
# 이미지 선택 및 플레이스홀더 생성
|
|
section_images = select_images_for_section(evidences, image_meta_by_file, MAX_IMAGES_PER_SECTION)
|
|
image_info_list = make_image_placeholders(it.number, section_images)
|
|
|
|
# ★ 추가: 이미지 파일을 assets 폴더로 복사
|
|
copy_images_to_assets(image_info_list)
|
|
copied_count = sum(1 for img in image_info_list if img.get('asset_path'))
|
|
total_images_copied += copied_count
|
|
|
|
# 사용자 프롬프트 생성
|
|
user_prompt = build_user_prompt(
|
|
report_title=report_title,
|
|
item=it,
|
|
evidences=evidences,
|
|
image_info_list=image_info_list,
|
|
previous_sections_summary=previous_sections_summary
|
|
)
|
|
|
|
# Gemini로 본문 생성
|
|
section_text = generate_section_text_gemini(system_instruction, user_prompt)
|
|
section_text = clean_generated_text_final(it.number, section_text) # ★ 이 한 줄만 추가!
|
|
|
|
# 마크다운 내용 추가
|
|
md_lines.append(f"\n#### {it.number} {it.title}\n")
|
|
md_lines.append(section_text + "\n")
|
|
|
|
# 중복 방지를 위해 현재 섹션 요약 누적 ← 이 부분은 그대로!
|
|
section_summary = extract_section_summary(section_text)
|
|
if section_summary:
|
|
previous_sections_summary += f"\n- {it.number}: {section_summary[:100]}..."
|
|
|
|
# JSON용 데이터 수집 (★ asset_path 추가)
|
|
report_json_sections.append({
|
|
"section_id": it.number,
|
|
"section_title": it.title,
|
|
"generated_text": section_text,
|
|
"sub_topics": [vars(st) for st in it.sub_topics],
|
|
"evidence_count": len(evidences),
|
|
"assets": [
|
|
{
|
|
"type": "image",
|
|
"image_id": img["image_id"],
|
|
"filename": img["image_file"],
|
|
"caption": img["caption"],
|
|
"placeholder": img["placeholder"],
|
|
"source_path": img.get("source_path", ""),
|
|
"page": img.get("page"),
|
|
"asset_path": img.get("asset_path"), # ★ 추가
|
|
}
|
|
for img in image_info_list
|
|
]
|
|
})
|
|
|
|
log(f" → 생성 완료 ({len(section_text)} 자)")
|
|
|
|
# 1. 마크다운(.md) 파일 저장
|
|
REPORT_MD_PATH.write_text("\n".join(md_lines), encoding="utf-8")
|
|
|
|
# 2. JSON(.json) 파일 저장
|
|
REPORT_JSON_PATH.write_text(
|
|
json.dumps({
|
|
"generated_at": datetime.now().isoformat(),
|
|
"report_title": report_title,
|
|
"model": GEMINI_MODEL,
|
|
"sections": report_json_sections
|
|
}, ensure_ascii=False, indent=2),
|
|
encoding="utf-8"
|
|
)
|
|
|
|
log(f"")
|
|
log(f"═══════════════════════════════════════════════════")
|
|
log(f"파일 저장 완료:")
|
|
log(f" 1. {REPORT_MD_PATH}")
|
|
log(f" 2. {REPORT_JSON_PATH}")
|
|
log(f" 3. {ASSETS_DIR} (이미지 {total_images_copied}개 복사)") # ★ 추가
|
|
log(f"═══════════════════════════════════════════════════")
|
|
log("=== step8 보고서 생성 종료 ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |