From d4a2e202c66b1a0c8b9ddb33e5e5937f51851f68 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9D=B4=EA=B2=BD=EB=AF=BC?= Date: Thu, 19 Mar 2026 14:01:42 +0900 Subject: [PATCH] =?UTF-8?q?Cleanup:=20Deleting=2003.Code/=EC=97=85?= =?UTF-8?q?=EB=A1=9C=EB=93=9C=EC=9A=A9/converters/pipeline/step8=5Fcontent?= =?UTF-8?q?.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../converters/pipeline/step8_content.py | 1021 ----------------- 1 file changed, 1021 deletions(-) delete mode 100644 03.Code/업로드용/converters/pipeline/step8_content.py diff --git a/03.Code/업로드용/converters/pipeline/step8_content.py b/03.Code/업로드용/converters/pipeline/step8_content.py deleted file mode 100644 index 0ea1cdb..0000000 --- a/03.Code/업로드용/converters/pipeline/step8_content.py +++ /dev/null @@ -1,1021 +0,0 @@ -# -*- 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\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 = 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()