diff --git a/03.Code/업로드용/converters/pipeline/step8_content.py b/03.Code/업로드용/converters/pipeline/step8_content.py index d2136b0..0ea1cdb 100644 --- a/03.Code/업로드용/converters/pipeline/step8_content.py +++ b/03.Code/업로드용/converters/pipeline/step8_content.py @@ -1,46 +1,104 @@ # -*- coding: utf-8 -*- -from dotenv import load_dotenv -load_dotenv() - """ step8_generate_report_gemini.py 기능 -- 생성된 목차(outline_issue_report.txt)를 읽어 각 섹션(소분류 목차)별 본문 내용을 생성합니다. -- 본문 생성 시 RAG 기술을 활용하여 코퍼스 및 원본 텍스트 조각을 근거로 제시합니다. -- google.genai 라이브러리를 사용하며, 템플릿에 따른 구조화된 출력을 생성합니다. +- 확정 목차(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 +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 + import faiss # type: ignore except Exception: faiss = None -# ===== API 설정 ===== +# ===== 하이브리드 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 = os.environ.get("OPENAI_API_KEY") +# 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 = os.environ.get("GEMINI_API_KEY") -GEMINI_MODEL = "gemini-3.1-pro-preview" +# 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\geulbeot-light\geulbeot-light\00.test\hwpx\out") +OUTPUT_ROOT = Path(r"D:\for python\geulbeot-light\geulbeot-light\00.test\hwpx\out\out") # 출력 위치 +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 = 10 -MAX_IMAGES_PER_SECTION = 3 +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}" @@ -48,6 +106,7 @@ def log(msg: str): with (LOG_DIR / "step8_generate_report_log.txt").open("a", encoding="utf-8") as f: f.write(line + "\n") + @dataclass class SubTopic: title: str @@ -55,6 +114,7 @@ class SubTopic: type: str guide: str + @dataclass class OutlineItem: number: str @@ -62,119 +122,900 @@ class OutlineItem: 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: - p = CONTEXT_DIR / "domain_prompt.txt" - if not p.exists(): - raise RuntimeError(f"domain_prompt.txt 없음: {p}") - return read_text(p) + 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]]: - p = CONTEXT_DIR / "outline_issue_report.txt" - if not p.exists(): - raise RuntimeError(f"목차 파일 없음: {p}") - raw = p.read_text(encoding="utf-8", errors="ignore").splitlines() + 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] = [] - - # 목차 파싱 로직 (step7 인덱스와 유사 구조) - # ... (중략: 상세 파싱 코드는 프로젝트 구조에 따라 최적화됨) + 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 retrieve_with_faiss(index, meta, item_map, query, top_k=10): - """FAISS 기반 근거 추출""" - resp = openai_client.embeddings.create(model="text-embedding-3-small", input=[query]) - q_vec = np.array(resp.data[0].embedding, dtype="float32").reshape(1, -1) - faiss.normalize_L2(q_vec) - - D, I = index.search(q_vec, top_k) - results = [] + +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: continue - m = meta[idx] - key = (m['source'], m['chunk']) - if key in item_map: - results.append(item_map[key]) - return results + 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: - return f""" -{domain_prompt} + """ + Gemini 시스템 지시문 (v4 - 최종) + """ + return f"""{domain_prompt} -당신은 보고서 작성 전문가입니다. 제공된 [근거 자료]를 바탕으로 논리적이고 전문적인 보고서 본문을 작성하십시오. -작성 시 주의사항: -1. 반드시 제공된 근거(수치, 기법, 명칭)에 기반하여 작성할 것. 근거가 없는 내용을 지어내지 마십시오. -2. 각 세부 주제별로 명확한 제목을 붙이고, 상세 내용을 기술하십시오. -3. 측량 분야의 용어와 기준을 정확하게 사용하십시오. -4. 문체는 "~함", "~임" 등 보고서 종결 어미를 사용하십시오. +═══════════════════════════════════════════════════════════════ + ★★★ 절대 준수 규칙 ★★★ +═══════════════════════════════════════════════════════════════ + +[금지 사항] +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 generate_section_text_gemini(system_instruction: str, user_prompt: str) -> str: + +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=0.3, # 낮은 temperature로 창의성 억제 ) ) return (response.text or "").strip() except Exception as e: log(f"[ERROR] Gemini API 호출 실패: {e}") - return f"[콘텐츠 생성 실패: {e}]" + return f"[생성 실패: {e}]" -def main(input_dir, output_dir, doc_type='report'): - global DATA_ROOT, OUTPUT_ROOT, CONTEXT_DIR, LOG_DIR, RAG_DIR, GEN_DIR - DATA_ROOT = Path(input_dir) - OUTPUT_ROOT = Path(output_dir) - CONTEXT_DIR = OUTPUT_ROOT / "context" - LOG_DIR = OUTPUT_ROOT / "logs" - RAG_DIR = OUTPUT_ROOT / "rag" - GEN_DIR = OUTPUT_ROOT / "generated" +import re + +def extract_section_summary(text: str, max_chars: int = 200) -> str: + """섹션 본문에서 핵심 키워드/주제 추출 (중복 방지용)""" + # 첫 200자 또는 첫 문단 + lines = text.split('\n') + summary_parts = [] + char_count = 0 - for d in [GEN_DIR, LOG_DIR]: - d.mkdir(parents=True, exist_ok=True) + 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] - log("=== 보고서 본문 생성 시작 (Gemini) ===") + +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() - # RAG 인덱스 로드 시도 - # ... (중략: FAISS 인덱스 로드 로직) + 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) - - final_report_md = [f"# {report_title}\n"] - - for item in outline_items: - if item.depth < 3: - final_report_md.append(f"\n{'#' * item.depth} {item.number} {item.title}\n") - continue - - log(f"처리 중: {item.number} {item.title}") - - # 근거 검색 및 본문 생성 - # query = f"{item.title} " + " ".join([st.title for st in item.sub_topics]) - # evidences = retrieve_with_faiss(...) - - user_prompt = f"섹션 '{item.number} {item.title}'에 대한 상세 보고서 내용을 작성하십시오.\n" - # user_prompt += "[근거 자료]\n..." - - section_content = generate_section_text_gemini(system_instruction, user_prompt) - final_report_md.append(f"\n#### {item.number} {item.title}\n") - final_report_md.append(section_content + "\n") - # 결과 저장 - report_path = GEN_DIR / "report_draft.md" - report_path.write_text("\n".join(final_report_md), encoding="utf-8") - log(f"보고서 초안 생성 완료: {report_path}") + 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()