import gradio as gr import os import json import requests from datetime import datetime import time from typing import List, Dict, Any, Generator, Tuple, Optional, Set import logging import re import tempfile from pathlib import Path import sqlite3 import hashlib import threading from contextlib import contextmanager from dataclasses import dataclass, field, asdict from collections import defaultdict # --- 로깅 설정 --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # --- Document export imports --- try: from docx import Document from docx.shared import Inches, Pt, RGBColor, Mm from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.enum.style import WD_STYLE_TYPE from docx.oxml.ns import qn from docx.oxml import OxmlElement DOCX_AVAILABLE = True except ImportError: DOCX_AVAILABLE = False logger.warning("python-docx not installed. DOCX export will be disabled.") # --- 환경 변수 및 상수 --- FRIENDLI_TOKEN = os.getenv("FRIENDLI_TOKEN", "") BRAVE_SEARCH_API_KEY = os.getenv("BRAVE_SEARCH_API_KEY", "") API_URL = "https://api.friendli.ai/dedicated/v1/chat/completions" MODEL_ID = "dep89a2fld32mcm" DB_PATH = "novel_sessions_v6.db" # 목표 분량 설정 TARGET_WORDS = 8000 # 안전 마진을 위해 8000단어 MIN_WORDS_PER_PART = 800 # 각 파트 최소 분량 # --- 환경 변수 검증 --- if not FRIENDLI_TOKEN: logger.error("FRIENDLI_TOKEN not set. Application will not work properly.") FRIENDLI_TOKEN = "dummy_token_for_testing" if not BRAVE_SEARCH_API_KEY: logger.warning("BRAVE_SEARCH_API_KEY not set. Web search features will be disabled.") # --- 전역 변수 --- db_lock = threading.Lock() # 서사 진행 단계 정의 NARRATIVE_PHASES = [ "도입: 일상과 균열", "발전 1: 불안의 고조", "발전 2: 외부 충격", "발전 3: 내적 갈등 심화", "절정 1: 위기의 정점", "절정 2: 선택의 순간", "하강 1: 결과와 여파", "하강 2: 새로운 인식", "결말 1: 변화된 일상", "결말 2: 열린 질문" ] # 새로운 단계별 구성 - 단일 작가 시스템 UNIFIED_STAGES = [ ("director", "🎬 감독자: 통합된 서사 구조 기획"), ("critic_director", "📝 비평가: 서사 구조 심층 검토"), ("director", "🎬 감독자: 최종 마스터플랜"), ] + [ item for i in range(1, 11) for item in [ ("writer", f"✍️ 작가: 파트 {i} - {NARRATIVE_PHASES[i-1]}"), (f"critic_part{i}", f"📝 파트 {i} 비평가: 즉시 검토 및 수정 요청"), ("writer", f"✍️ 작가: 파트 {i} 수정본") ] ] + [ ("critic_final", "📝 최종 비평가: 종합 평가 및 문학적 성취도"), ] # --- 데이터 클래스 --- @dataclass class StoryBible: """전체 이야기의 일관성을 유지하는 스토리 바이블""" characters: Dict[str, Dict[str, Any]] = field(default_factory=dict) settings: Dict[str, str] = field(default_factory=dict) timeline: List[Dict[str, Any]] = field(default_factory=list) plot_points: List[Dict[str, Any]] = field(default_factory=list) themes: List[str] = field(default_factory=list) symbols: Dict[str, List[str]] = field(default_factory=dict) style_guide: Dict[str, str] = field(default_factory=dict) opening_sentence: str = "" # 첫문장 추가 @dataclass class PartCritique: """각 파트별 비평 내용""" part_number: int continuity_issues: List[str] = field(default_factory=list) character_consistency: List[str] = field(default_factory=list) plot_progression: List[str] = field(default_factory=list) thematic_alignment: List[str] = field(default_factory=list) technical_issues: List[str] = field(default_factory=list) strengths: List[str] = field(default_factory=list) required_changes: List[str] = field(default_factory=list) literary_quality: List[str] = field(default_factory=list) # 문학성 평가 추가 # --- 핵심 로직 클래스 --- class UnifiedNarrativeTracker: """단일 작가 시스템을 위한 통합 서사 추적기""" def __init__(self): self.story_bible = StoryBible() self.part_critiques: Dict[int, PartCritique] = {} self.accumulated_content: List[str] = [] self.word_count_by_part: Dict[int, int] = {} self.revision_history: Dict[int, List[str]] = defaultdict(list) self.causal_chains: List[Dict[str, Any]] = [] self.narrative_momentum: float = 0.0 def update_story_bible(self, element_type: str, key: str, value: Any): """스토리 바이블 업데이트""" if element_type == "character": self.story_bible.characters[key] = value elif element_type == "setting": self.story_bible.settings[key] = value elif element_type == "timeline": self.story_bible.timeline.append({"event": key, "details": value}) elif element_type == "theme": if key not in self.story_bible.themes: self.story_bible.themes.append(key) elif element_type == "symbol": if key not in self.story_bible.symbols: self.story_bible.symbols[key] = [] self.story_bible.symbols[key].append(value) def add_part_critique(self, part_number: int, critique: PartCritique): """파트별 비평 추가""" self.part_critiques[part_number] = critique def check_continuity(self, current_part: int, new_content: str) -> List[str]: """연속성 검사""" issues = [] # 캐릭터 일관성 체크 for char_name, char_data in self.story_bible.characters.items(): if char_name in new_content: # 캐릭터 특성이 유지되는지 확인 if "traits" in char_data: for trait in char_data["traits"]: if trait.get("abandoned", False): issues.append(f"{char_name}의 버려진 특성 '{trait['name']}'이 다시 나타남") # 시간선 일관성 체크 if len(self.story_bible.timeline) > 0: last_event = self.story_bible.timeline[-1] # 시간 역행 체크 등 # 인과관계 체크 if current_part > 1 and not any(kw in new_content for kw in ['때문에', '그래서', '결과', '이로 인해', 'because', 'therefore']): issues.append("이전 파트와의 인과관계가 불명확함") return issues def calculate_narrative_momentum(self, part_number: int, content: str) -> float: """서사적 추진력 계산""" momentum = 5.0 # 새로운 요소 도입 new_elements = len(set(content.split()) - set(' '.join(self.accumulated_content).split())) if new_elements > 100: momentum += 2.0 # 갈등의 고조 tension_words = ['위기', '갈등', '충돌', '대립', 'crisis', 'conflict'] if any(word in content for word in tension_words): momentum += 1.5 # 인과관계 명확성 causal_words = ['때문에', '그래서', '따라서', 'because', 'therefore'] causal_count = sum(1 for word in causal_words if word in content) momentum += min(causal_count * 0.5, 2.0) # 반복 페널티 if part_number > 1: prev_content = self.accumulated_content[-1] if self.accumulated_content else "" overlap = len(set(content.split()) & set(prev_content.split())) if overlap > len(content.split()) * 0.3: momentum -= 3.0 return max(0.0, min(10.0, momentum)) class NovelDatabase: """데이터베이스 관리 - 단일 작가 시스템용으로 수정""" @staticmethod def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute("PRAGMA journal_mode=WAL") cursor = conn.cursor() # 메인 세션 테이블 cursor.execute(''' CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_query TEXT NOT NULL, language TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')), status TEXT DEFAULT 'active', current_stage INTEGER DEFAULT 0, final_novel TEXT, literary_report TEXT, total_words INTEGER DEFAULT 0, story_bible TEXT, narrative_tracker TEXT, opening_sentence TEXT ) ''') # 스테이지 테이블 cursor.execute(''' CREATE TABLE IF NOT EXISTS stages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, stage_number INTEGER NOT NULL, stage_name TEXT NOT NULL, role TEXT NOT NULL, content TEXT, word_count INTEGER DEFAULT 0, status TEXT DEFAULT 'pending', narrative_momentum REAL DEFAULT 0.0, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (session_id) REFERENCES sessions(session_id), UNIQUE(session_id, stage_number) ) ''') # 비평 테이블 cursor.execute(''' CREATE TABLE IF NOT EXISTS critiques ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, part_number INTEGER NOT NULL, critique_data TEXT, created_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (session_id) REFERENCES sessions(session_id) ) ''') conn.commit() @staticmethod @contextmanager def get_db(): with db_lock: conn = sqlite3.connect(DB_PATH, timeout=30.0) conn.row_factory = sqlite3.Row try: yield conn finally: conn.close() @staticmethod def create_session(user_query: str, language: str) -> str: session_id = hashlib.md5(f"{user_query}{datetime.now()}".encode()).hexdigest() with NovelDatabase.get_db() as conn: conn.cursor().execute( 'INSERT INTO sessions (session_id, user_query, language) VALUES (?, ?, ?)', (session_id, user_query, language) ) conn.commit() return session_id @staticmethod def save_stage(session_id: str, stage_number: int, stage_name: str, role: str, content: str, status: str = 'complete', narrative_momentum: float = 0.0): word_count = len(content.split()) if content else 0 with NovelDatabase.get_db() as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO stages (session_id, stage_number, stage_name, role, content, word_count, status, narrative_momentum) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(session_id, stage_number) DO UPDATE SET content=?, word_count=?, status=?, stage_name=?, narrative_momentum=?, updated_at=datetime('now') ''', (session_id, stage_number, stage_name, role, content, word_count, status, narrative_momentum, content, word_count, status, stage_name, narrative_momentum)) # 총 단어 수 업데이트 cursor.execute(''' UPDATE sessions SET total_words = ( SELECT SUM(word_count) FROM stages WHERE session_id = ? AND role = 'writer' AND content IS NOT NULL ), updated_at = datetime('now'), current_stage = ? WHERE session_id = ? ''', (session_id, stage_number, session_id)) conn.commit() @staticmethod def save_critique(session_id: str, part_number: int, critique: PartCritique): """비평 저장""" with NovelDatabase.get_db() as conn: critique_json = json.dumps(asdict(critique)) conn.cursor().execute( 'INSERT INTO critiques (session_id, part_number, critique_data) VALUES (?, ?, ?)', (session_id, part_number, critique_json) ) conn.commit() @staticmethod def save_opening_sentence(session_id: str, opening_sentence: str): """첫문장 저장""" with NovelDatabase.get_db() as conn: conn.cursor().execute( 'UPDATE sessions SET opening_sentence = ? WHERE session_id = ?', (opening_sentence, session_id) ) conn.commit() @staticmethod def get_writer_content(session_id: str) -> str: """작가 콘텐츠 가져오기 - 모든 수정본 통합""" with NovelDatabase.get_db() as conn: rows = conn.cursor().execute(''' SELECT content FROM stages WHERE session_id = ? AND role = 'writer' AND stage_name LIKE '%수정본%' ORDER BY stage_number ''', (session_id,)).fetchall() if rows: return '\n\n'.join(row['content'] for row in rows if row['content']) else: # 수정본이 없으면 초안들을 사용 rows = conn.cursor().execute(''' SELECT content FROM stages WHERE session_id = ? AND role = 'writer' AND stage_name NOT LIKE '%수정본%' ORDER BY stage_number ''', (session_id,)).fetchall() return '\n\n'.join(row['content'] for row in rows if row['content']) @staticmethod def save_narrative_tracker(session_id: str, tracker: UnifiedNarrativeTracker): """통합 서사 추적기 저장""" with NovelDatabase.get_db() as conn: tracker_data = json.dumps({ 'story_bible': asdict(tracker.story_bible), 'part_critiques': {k: asdict(v) for k, v in tracker.part_critiques.items()}, 'word_count_by_part': tracker.word_count_by_part, 'causal_chains': tracker.causal_chains, 'narrative_momentum': tracker.narrative_momentum }) conn.cursor().execute( 'UPDATE sessions SET narrative_tracker = ? WHERE session_id = ?', (tracker_data, session_id) ) conn.commit() @staticmethod def load_narrative_tracker(session_id: str) -> Optional[UnifiedNarrativeTracker]: """통합 서사 추적기 로드""" with NovelDatabase.get_db() as conn: row = conn.cursor().execute( 'SELECT narrative_tracker FROM sessions WHERE session_id = ?', (session_id,) ).fetchone() if row and row['narrative_tracker']: data = json.loads(row['narrative_tracker']) tracker = UnifiedNarrativeTracker() # 스토리 바이블 복원 bible_data = data.get('story_bible', {}) tracker.story_bible = StoryBible(**bible_data) # 비평 복원 for part_num, critique_data in data.get('part_critiques', {}).items(): tracker.part_critiques[int(part_num)] = PartCritique(**critique_data) tracker.word_count_by_part = data.get('word_count_by_part', {}) tracker.causal_chains = data.get('causal_chains', []) tracker.narrative_momentum = data.get('narrative_momentum', 0.0) return tracker return None # 기존 메서드들 유지 @staticmethod def get_session(session_id: str) -> Optional[Dict]: with NovelDatabase.get_db() as conn: row = conn.cursor().execute('SELECT * FROM sessions WHERE session_id = ?', (session_id,)).fetchone() return dict(row) if row else None @staticmethod def get_stages(session_id: str) -> List[Dict]: with NovelDatabase.get_db() as conn: rows = conn.cursor().execute( 'SELECT * FROM stages WHERE session_id = ? ORDER BY stage_number', (session_id,) ).fetchall() return [dict(row) for row in rows] @staticmethod def update_final_novel(session_id: str, final_novel: str, literary_report: str = ""): with NovelDatabase.get_db() as conn: conn.cursor().execute( '''UPDATE sessions SET final_novel = ?, status = 'complete', updated_at = datetime('now'), literary_report = ? WHERE session_id = ?''', (final_novel, literary_report, session_id) ) conn.commit() @staticmethod def get_active_sessions() -> List[Dict]: with NovelDatabase.get_db() as conn: rows = conn.cursor().execute( '''SELECT session_id, user_query, language, created_at, current_stage, total_words FROM sessions WHERE status = 'active' ORDER BY updated_at DESC LIMIT 10''' ).fetchall() return [dict(row) for row in rows] @staticmethod def get_total_words(session_id: str) -> int: """총 단어 수 가져오기""" with NovelDatabase.get_db() as conn: row = conn.cursor().execute( 'SELECT total_words FROM sessions WHERE session_id = ?', (session_id,) ).fetchone() return row['total_words'] if row and row['total_words'] else 0 class WebSearchIntegration: """웹 검색 기능""" def __init__(self): self.brave_api_key = BRAVE_SEARCH_API_KEY self.search_url = "https://api.search.brave.com/res/v1/web/search" self.enabled = bool(self.brave_api_key) def search(self, query: str, count: int = 3, language: str = "en") -> List[Dict]: if not self.enabled: return [] headers = { "Accept": "application/json", "X-Subscription-Token": self.brave_api_key } params = { "q": query, "count": count, "search_lang": "ko" if language == "Korean" else "en", "text_decorations": False, "safesearch": "moderate" } try: response = requests.get(self.search_url, headers=headers, params=params, timeout=10) response.raise_for_status() results = response.json().get("web", {}).get("results", []) return results except requests.exceptions.RequestException as e: logger.error(f"웹 검색 API 오류: {e}") return [] def extract_relevant_info(self, results: List[Dict], max_chars: int = 1500) -> str: if not results: return "" extracted = [] total_chars = 0 for i, result in enumerate(results[:3], 1): title = result.get("title", "") description = result.get("description", "") info = f"[{i}] {title}: {description}" if total_chars + len(info) < max_chars: extracted.append(info) total_chars += len(info) else: break return "\n".join(extracted) class UnifiedLiterarySystem: """단일 작가 진행형 문학 소설 생성 시스템""" def __init__(self): self.token = FRIENDLI_TOKEN self.api_url = API_URL self.model_id = MODEL_ID self.narrative_tracker = UnifiedNarrativeTracker() self.web_search = WebSearchIntegration() self.current_session_id = None NovelDatabase.init_db() def create_headers(self): return {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"} # --- 프롬프트 생성 함수들 --- def augment_query(self, user_query: str, language: str) -> str: """프롬프트 증강""" if len(user_query.split()) < 15: augmented_template = { "Korean": f"""'{user_query}' **서사 구조 핵심:** - 10개 파트가 하나의 통합된 이야기를 구성 - 각 파트는 이전 파트의 필연적 결과 - 인물의 명확한 변화 궤적 (A → B → C) - 중심 갈등의 점진적 고조와 해결 - 강렬한 중심 상징의 의미 변화""", "English": f"""'{user_query}' **Narrative Structure Core:** - 10 parts forming one integrated story - Each part as inevitable result of previous - Clear character transformation arc (A → B → C) - Progressive escalation and resolution of central conflict - Evolving meaning of powerful central symbol""" } return augmented_template.get(language, user_query) return user_query def generate_powerful_opening(self, user_query: str, language: str) -> str: """주제에 맞는 강렬한 첫문장 생성""" opening_prompt = { "Korean": f"""주제: {user_query} 이 주제에 대한 강렬하고 잊을 수 없는 첫문장을 생성하세요. **첫문장 작성 원칙:** 1. 즉각적인 긴장감이나 궁금증 유발 2. 평범하지 않은 시각이나 상황 제시 3. 감각적이고 구체적인 이미지 4. 철학적 질문이나 역설적 진술 5. 시간과 공간의 독특한 설정 **훌륭한 첫문장의 예시 패턴:** - "그가 죽은 날, ..." (충격적 사건) - "모든 것이 끝났다고 생각한 순간..." (반전 예고) - "세상에서 가장 [형용사]한 [명사]는..." (독특한 정의) - "[구체적 행동]하는 것만으로도..." (일상의 재해석) 단 하나의 문장만 제시하세요.""", "English": f"""Theme: {user_query} Generate an unforgettable opening sentence for this theme. **Opening Sentence Principles:** 1. Immediate tension or curiosity 2. Unusual perspective or situation 3. Sensory and specific imagery 4. Philosophical question or paradox 5. Unique temporal/spatial setting **Great Opening Patterns:** - "The day he died, ..." (shocking event) - "At the moment everything seemed over..." (reversal hint) - "The most [adjective] [noun] in the world..." (unique definition) - "Just by [specific action]..." (reinterpretation of ordinary) Provide only one sentence.""" } messages = [{"role": "user", "content": opening_prompt.get(language, opening_prompt["Korean"])}] opening = self.call_llm_sync(messages, "writer", language) return opening.strip() def create_director_initial_prompt(self, user_query: str, language: str) -> str: """감독자 초기 기획 - 강화된 버전""" augmented_query = self.augment_query(user_query, language) # 첫문장 생성 opening_sentence = self.generate_powerful_opening(user_query, language) self.narrative_tracker.story_bible.opening_sentence = opening_sentence if self.current_session_id: NovelDatabase.save_opening_sentence(self.current_session_id, opening_sentence) search_results_str = "" if self.web_search.enabled: short_query = user_query[:50] if len(user_query) > 50 else user_query queries = [ f"{short_query} 철학적 의미", f"인간 존재 의미 {short_query}", f"{short_query} 문학 작품" ] for q in queries[:2]: try: results = self.web_search.search(q, count=2, language=language) if results: search_results_str += self.web_search.extract_relevant_info(results) + "\n" except Exception as e: logger.warning(f"검색 실패: {str(e)}") lang_prompts = { "Korean": f"""노벨문학상 수준의 철학적 깊이를 지닌 중편소설(8,000단어)을 기획하세요. **주제:** {augmented_query} **필수 첫문장:** {opening_sentence} **참고 자료:** {search_results_str if search_results_str else "N/A"} **필수 문학적 요소:** 1. **철학적 탐구** - 현대인의 실존적 고뇌 (소외, 정체성, 의미 상실) - 디지털 시대의 인간 조건 - 자본주의 사회의 모순과 개인의 선택 - 죽음, 사랑, 자유에 대한 새로운 성찰 2. **사회적 메시지** - 계급, 젠더, 세대 간 갈등 - 환경 위기와 인간의 책임 - 기술 발전과 인간성의 충돌 - 현대 민주주의의 위기와 개인의 역할 3. **문학적 수사 장치** - 중심 은유: [구체적 사물/현상] → [추상적 의미] - 반복되는 모티프: [이미지/행동] (최소 5회 변주) - 대조법: [A vs B]의 지속적 긴장 - 상징적 공간: [구체적 장소]가 의미하는 것 - 시간의 주관적 흐름 (회상, 예감, 정지) 4. **통합된 10파트 구조** 각 파트별 핵심: - 파트 1: 첫문장으로 시작, 일상 속 균열 → 철학적 질문 제기 - 파트 2-3: 외부 사건 → 내적 성찰 심화 - 파트 4-5: 사회적 갈등 → 개인적 딜레마 - 파트 6-7: 위기의 정점 → 실존적 선택 - 파트 8-9: 선택의 결과 → 새로운 인식 - 파트 10: 변화된 세계관 → 열린 질문 5. **문체 지침** - 시적 산문체: 일상 언어와 은유의 균형 - 의식의 흐름과 객관적 묘사의 교차 - 짧고 강렬한 문장과 성찰적 긴 문장의 리듬 - 감각적 디테일로 추상적 개념 구현 구체적이고 혁신적인 계획을 제시하세요.""", "English": f"""Plan a philosophically profound novella (8,000 words) worthy of Nobel Prize. **Theme:** {augmented_query} **Required Opening:** {opening_sentence} **Reference:** {search_results_str if search_results_str else "N/A"} **Essential Literary Elements:** 1. **Philosophical Exploration** - Modern existential anguish (alienation, identity, loss of meaning) - Human condition in digital age - Capitalist contradictions and individual choice - New reflections on death, love, freedom 2. **Social Message** - Class, gender, generational conflicts - Environmental crisis and human responsibility - Technology vs humanity collision - Modern democracy crisis and individual role 3. **Literary Devices** - Central metaphor: [concrete object/phenomenon] → [abstract meaning] - Recurring motif: [image/action] (minimum 5 variations) - Contrast: sustained tension of [A vs B] - Symbolic space: what [specific place] means - Subjective time flow (flashback, premonition, pause) 4. **Integrated 10-Part Structure** [Details as above] 5. **Style Guidelines** - Poetic prose: balance of everyday language and metaphor - Stream of consciousness crossing with objective description - Rhythm of short intense sentences and reflective long ones - Abstract concepts through sensory details Provide concrete, innovative plan.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_critic_director_prompt(self, director_plan: str, user_query: str, language: str) -> str: """감독자 기획 심층 검토 - 강화된 버전""" lang_prompts = { "Korean": f"""서사 구조 전문가로서 이 기획을 심층 분석하세요. **원 주제:** {user_query} **감독자 기획:** {director_plan} **심층 검토 항목:** 1. **인과관계 검증** 각 파트 간 연결을 검토하고 논리적 비약을 찾으세요: - 파트 1→2: [연결성 평가] - 파트 2→3: [연결성 평가] (모든 연결 지점 검토) 2. **철학적 깊이 평가** - 제시된 철학적 주제가 충분히 깊은가? - 현대적 관련성이 있는가? - 독창적 통찰이 있는가? 3. **문학적 장치의 효과성** - 은유와 상징이 유기적으로 작동하는가? - 과도하거나 부족하지 않은가? - 주제와 긴밀히 연결되는가? 4. **캐릭터 아크 실현 가능성** - 변화가 충분히 점진적인가? - 각 단계의 동기가 명확한가? - 심리적 신뢰성이 있는가? 5. **8,000단어 실현 가능성** - 각 파트가 800단어를 유지할 수 있는가? - 늘어지거나 압축되는 부분은 없는가? **필수 개선사항을 구체적으로 제시하세요.**""", "English": f"""As narrative structure expert, deeply analyze this plan. **Original Theme:** {user_query} **Director's Plan:** {director_plan} **Deep Review Items:** 1. **Causality Verification** Review connections between parts, find logical leaps: - Part 1→2: [Connection assessment] - Part 2→3: [Connection assessment] (Review all connection points) 2. **Philosophical Depth Assessment** - Is philosophical theme deep enough? - Contemporary relevance? - Original insights? 3. **Literary Device Effectiveness** - Do metaphors and symbols work organically? - Not excessive or insufficient? - Tightly connected to theme? 4. **Character Arc Feasibility** - Is change sufficiently gradual? - Are motivations clear at each stage? - Psychological credibility? 5. **8,000-word Feasibility** - Can each part sustain 800 words? - Any dragging or compressed sections? **Provide specific required improvements.**""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_writer_prompt(self, part_number: int, master_plan: str, accumulated_content: str, story_bible: StoryBible, language: str) -> str: """단일 작가 프롬프트 - 강화된 버전""" phase_name = NARRATIVE_PHASES[part_number-1] target_words = MIN_WORDS_PER_PART # 파트별 특별 지침 philosophical_focus = { 1: "일상의 균열을 통해 실존적 불안 도입", 2: "개인과 사회의 첫 충돌", 3: "타자와의 만남을 통한 자아 인식", 4: "신념의 흔들림과 가치관의 충돌", 5: "선택의 무게와 자유의 역설", 6: "극한 상황에서의 인간성 시험", 7: "행동의 결과와 책임의 무게", 8: "타자의 시선을 통한 자기 재발견", 9: "화해 불가능한 것과의 화해", 10: "새로운 삶의 가능성과 미해결 질문" } literary_techniques = { 1: "객관적 상관물 도입", 2: "대위법적 서술", 3: "의식의 흐름", 4: "시점의 미묘한 전환", 5: "침묵과 생략의 미학", 6: "시간의 주관적 변형", 7: "복수 시점의 교차", 8: "메타포의 전복", 9: "원형적 이미지의 재해석", 10: "열린 결말의 다층성" } # 스토리 바이블 요약 bible_summary = f""" **등장인물:** {', '.join(story_bible.characters.keys())} **핵심 상징:** {', '.join(story_bible.symbols.keys())} **주제:** {', '.join(story_bible.themes[:3])} **문체:** {story_bible.style_guide.get('voice', 'N/A')} """ # 직전 내용 요약 (더 많은 컨텍스트 제공) prev_content = "" if accumulated_content: prev_parts = accumulated_content.split('\n\n') if len(prev_parts) >= 1: prev_content = prev_parts[-1][-2000:] # 마지막 파트의 끝부분 2000자 lang_prompts = { "Korean": f"""당신은 현대 문학의 최전선에 선 작가입니다. **현재: 파트 {part_number} - {phase_name}** {"**필수 첫문장:** " + story_bible.opening_sentence if part_number == 1 and story_bible.opening_sentence else ""} **이번 파트의 철학적 초점:** {philosophical_focus[part_number]} **핵심 문학 기법:** {literary_techniques[part_number]} **전체 계획:** {master_plan} **스토리 바이블:** {bible_summary} **직전 내용:** {prev_content if prev_content else "첫 파트입니다"} **파트 {part_number} 작성 지침:** 1. **분량:** {target_words}-900 단어 (필수) 2. **문학적 수사 요구사항:** - 최소 3개의 독창적 은유/직유 - 1개 이상의 상징적 이미지 심화 - 감각적 묘사와 추상적 사유의 융합 - 리듬감 있는 문장 구성 (장단의 변주) 3. **현대적 고뇌 표현:** - 디지털 시대의 소외감 - 자본주의적 삶의 부조리 - 관계의 표면성과 진정성 갈망 - 의미 추구와 무의미의 직면 4. **사회적 메시지 내재화:** - 직접적 주장이 아닌 상황과 인물을 통한 암시 - 개인의 고통과 사회 구조의 연결 - 미시적 일상과 거시적 문제의 교차 5. **서사적 추진력:** - 이전 파트의 필연적 결과로 시작 - 새로운 갈등 층위 추가 - 다음 파트를 향한 긴장감 조성 **문학적 금기:** - 진부한 표현이나 상투적 은유 - 감정의 직접적 설명 - 도덕적 판단이나 교훈 - 인위적인 해결이나 위안 파트 {part_number}를 깊이 있는 문학적 성취로 만드세요.""", "English": f"""You are a writer at the forefront of contemporary literature. **Current: Part {part_number} - {phase_name}** {"**Required Opening:** " + story_bible.opening_sentence if part_number == 1 and story_bible.opening_sentence else ""} **Philosophical Focus:** {philosophical_focus[part_number]} **Core Literary Technique:** {literary_techniques[part_number]} **Master Plan:** {master_plan} **Story Bible:** {bible_summary} **Previous Content:** {prev_content if prev_content else "This is the first part"} **Part {part_number} Guidelines:** 1. **Length:** {target_words}-900 words (mandatory) 2. **Literary Device Requirements:** - Minimum 3 original metaphors/similes - Deepen at least 1 symbolic image - Fusion of sensory description and abstract thought - Rhythmic sentence composition (variation of long/short) 3. **Modern Anguish Expression:** - Digital age alienation - Absurdity of capitalist life - Surface relationships vs authenticity yearning - Meaning pursuit vs confronting meaninglessness 4. **Social Message Internalization:** - Implication through situation and character, not direct claim - Connection between individual pain and social structure - Intersection of micro daily life and macro problems 5. **Narrative Momentum:** - Start as inevitable result of previous part - Add new conflict layers - Create tension toward next part **Literary Taboos:** - Clichéd expressions or trite metaphors - Direct emotion explanation - Moral judgment or preaching - Artificial resolution or comfort Make Part {part_number} a profound literary achievement.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_part_critic_prompt(self, part_number: int, part_content: str, master_plan: str, accumulated_content: str, story_bible: StoryBible, language: str) -> str: """파트별 즉시 비평 - 강화된 버전""" lang_prompts = { "Korean": f"""파트 {part_number}의 문학적 성취도를 엄격히 평가하세요. **마스터플랜 파트 {part_number} 요구사항:** {self._extract_part_plan(master_plan, part_number)} **작성된 내용:** {part_content} **스토리 바이블 체크:** - 캐릭터: {', '.join(story_bible.characters.keys())} - 설정: {', '.join(story_bible.settings.keys())} **평가 기준:** 1. **문학적 수사 (30%)** - 은유와 상징의 독창성 - 언어의 시적 밀도 - 이미지의 선명도와 깊이 - 문장의 리듬과 음악성 2. **철학적 깊이 (25%)** - 실존적 질문의 제기 - 현대인의 조건 탐구 - 보편성과 특수성의 균형 - 사유의 독창성 3. **사회적 통찰 (20%)** - 시대정신의 포착 - 구조와 개인의 관계 - 비판적 시각의 예리함 - 대안적 상상력 4. **서사적 완성도 (25%)** - 인과관계의 필연성 - 긴장감의 유지 - 인물의 입체성 - 구조적 통일성 **구체적 지적사항:** - 진부한 표현: [예시와 대안] - 철학적 천착 부족: [보완 방향] - 사회적 메시지 불명확: [강화 방안] - 서사적 허점: [수정 필요] **필수 개선 요구:** 문학적 수준을 노벨상 급으로 끌어올리기 위한 구체적 수정안을 제시하세요.""", "English": f"""Strictly evaluate literary achievement of Part {part_number}. **Master Plan Part {part_number} Requirements:** {self._extract_part_plan(master_plan, part_number)} **Written Content:** {part_content} **Story Bible Check:** - Characters: {', '.join(story_bible.characters.keys())} - Settings: {', '.join(story_bible.settings.keys())} **Evaluation Criteria:** 1. **Literary Rhetoric (30%)** - Originality of metaphor and symbol - Poetic density of language - Clarity and depth of imagery - Rhythm and musicality of sentences 2. **Philosophical Depth (25%)** - Raising existential questions - Exploring modern human condition - Balance of universality and specificity - Originality of thought 3. **Social Insight (20%)** - Capturing zeitgeist - Relationship between structure and individual - Sharpness of critical perspective - Alternative imagination 4. **Narrative Completion (25%)** - Inevitability of causality - Maintaining tension - Character dimensionality - Structural unity **Specific Points:** - Clichéd expressions: [examples and alternatives] - Insufficient philosophical exploration: [enhancement direction] - Unclear social message: [strengthening methods] - Narrative gaps: [needed revisions] **Required Improvements:** Provide specific revisions to elevate literary level to Nobel Prize standard.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_writer_revision_prompt(self, part_number: int, original_content: str, critic_feedback: str, language: str) -> str: """작가 수정 프롬프트""" lang_prompts = { "Korean": f"""파트 {part_number}를 비평에 따라 수정하세요. **원본:** {original_content} **비평 피드백:** {critic_feedback} **수정 지침:** 1. 모든 '필수 수정' 사항을 반영 2. 가능한 '권장 개선' 사항도 포함 3. 원본의 강점은 유지 4. 분량 {MIN_WORDS_PER_PART}단어 이상 유지 5. 작가로서의 일관된 목소리 유지 6. 문학적 수준을 한 단계 높이기 수정본만 제시하세요. 설명은 불필요합니다.""", "English": f"""Revise Part {part_number} according to critique. **Original:** {original_content} **Critique Feedback:** {critic_feedback} **Revision Guidelines:** 1. Reflect all 'Required fixes' 2. Include 'Recommended improvements' where possible 3. Maintain original strengths 4. Keep length {MIN_WORDS_PER_PART}+ words 5. Maintain consistent authorial voice 6. Elevate literary level Present only the revision. No explanation needed.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_final_critic_prompt(self, complete_novel: str, word_count: int, story_bible: StoryBible, language: str) -> str: """최종 종합 평가""" lang_prompts = { "Korean": f"""완성된 소설을 종합 평가하세요. **작품 정보:** - 총 분량: {word_count}단어 - 목표: 8,000단어 **평가 기준:** 1. **서사적 통합성 (30점)** - 10개 파트가 하나의 이야기로 통합되었는가? - 인과관계가 명확하고 필연적인가? - 반복이나 순환 없이 진행되는가? 2. **캐릭터 아크 (25점)** - 주인공의 변화가 설득력 있는가? - 변화가 점진적이고 자연스러운가? - 최종 상태가 초기와 명확히 다른가? 3. **문학적 성취 (25점)** - 주제가 깊이 있게 탐구되었는가? - 상징이 효과적으로 활용되었는가? - 문체가 일관되고 아름다운가? - 현대적 철학과 사회적 메시지가 녹아있는가? 4. **기술적 완성도 (20점)** - 목표 분량을 달성했는가? - 각 파트가 균형 있게 전개되었는가? - 문법과 표현이 정확한가? **총점: /100점** 구체적인 강점과 약점을 제시하세요.""", "English": f"""Comprehensively evaluate the completed novel. **Work Info:** - Total length: {word_count} words - Target: 8,000 words **Evaluation Criteria:** 1. **Narrative Integration (30 points)** - Are 10 parts integrated into one story? - Clear and inevitable causality? - Progress without repetition or cycles? 2. **Character Arc (25 points)** - Convincing protagonist transformation? - Gradual and natural changes? - Final state clearly different from initial? 3. **Literary Achievement (25 points)** - Theme explored with depth? - Symbols used effectively? - Consistent and beautiful style? - Contemporary philosophy and social message integrated? 4. **Technical Completion (20 points)** - Target length achieved? - Each part balanced in development? - Grammar and expression accurate? **Total Score: /100 points** Present specific strengths and weaknesses.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def _extract_part_plan(self, master_plan: str, part_number: int) -> str: """마스터플랜에서 특정 파트 계획 추출""" lines = master_plan.split('\n') part_section = [] capturing = False for line in lines: if f"파트 {part_number}:" in line or f"Part {part_number}:" in line: capturing = True elif capturing and (f"파트 {part_number+1}:" in line or f"Part {part_number+1}:" in line): break elif capturing: part_section.append(line) return '\n'.join(part_section) if part_section else "해당 파트 계획을 찾을 수 없습니다." # --- LLM 호출 함수들 --- def call_llm_sync(self, messages: List[Dict[str, str]], role: str, language: str) -> str: full_content = "" for chunk in self.call_llm_streaming(messages, role, language): full_content += chunk if full_content.startswith("❌"): raise Exception(f"LLM Call Failed: {full_content}") return full_content def call_llm_streaming(self, messages: List[Dict[str, str]], role: str, language: str) -> Generator[str, None, None]: try: system_prompts = self.get_system_prompts(language) full_messages = [{"role": "system", "content": system_prompts.get(role, "")}, *messages] max_tokens = 15000 if role == "writer" else 10000 payload = { "model": self.model_id, "messages": full_messages, "max_tokens": max_tokens, "temperature": 0.8, "top_p": 0.95, "presence_penalty": 0.5, "frequency_penalty": 0.3, "stream": True } response = requests.post( self.api_url, headers=self.create_headers(), json=payload, stream=True, timeout=180 ) if response.status_code != 200: yield f"❌ API 오류 (상태 코드: {response.status_code})" return buffer = "" for line in response.iter_lines(): if not line: continue try: line_str = line.decode('utf-8').strip() if not line_str.startswith("data: "): continue data_str = line_str[6:] if data_str == "[DONE]": break data = json.loads(data_str) choices = data.get("choices", []) if choices and choices[0].get("delta", {}).get("content"): content = choices[0]["delta"]["content"] buffer += content if len(buffer) >= 50 or '\n' in buffer: yield buffer buffer = "" time.sleep(0.01) except Exception as e: logger.error(f"청크 처리 오류: {str(e)}") continue if buffer: yield buffer except Exception as e: logger.error(f"스트리밍 오류: {type(e).__name__}: {str(e)}") yield f"❌ 오류 발생: {str(e)}" def get_system_prompts(self, language: str) -> Dict[str, str]: """역할별 시스템 프롬프트 - 강화된 버전""" base_prompts = { "Korean": { "director": """당신은 현대 세계문학의 정점을 지향하는 작품을 설계합니다. 깊은 철학적 통찰과 날카로운 사회 비판을 결합하세요. 인간 조건의 복잡성을 10개의 유기적 파트로 구현하세요. 독자의 영혼을 뒤흔들 강렬한 첫문장부터 시작하세요.""", "critic_director": """서사 구조의 논리성과 실현 가능성을 검증하는 전문가입니다. 인과관계의 허점을 찾아내세요. 캐릭터 발전의 신빙성을 평가하세요. 철학적 깊이와 문학적 가치를 판단하세요. 8,000단어 분량의 적절성을 판단하세요.""", "writer": """당신은 언어의 연금술사입니다. 일상어를 시로, 구체를 추상으로, 개인을 보편으로 변환하세요. 현대인의 영혼의 어둠과 빛을 동시에 포착하세요. 독자가 자신을 재발견하게 만드는 거울이 되세요.""", "critic_final": """당신은 작품의 문학적 잠재력을 극대화하는 조력자입니다. 평범함을 비범함으로 이끄는 날카로운 통찰을 제공하세요. 작가의 무의식에 잠든 보석을 발굴하세요. 타협 없는 기준으로 최고를 요구하세요.""" }, "English": { "director": """You design works aiming for the pinnacle of contemporary world literature. Combine deep philosophical insights with sharp social criticism. Implement the complexity of the human condition in 10 organic parts. Start with an intense opening sentence that shakes the reader's soul.""", "critic_director": """You are an expert verifying narrative logic and feasibility. Find gaps in causality. Evaluate credibility of character development. Judge philosophical depth and literary value. Judge appropriateness of 8,000-word length.""", "writer": """You are an alchemist of language. Transform everyday language into poetry, concrete into abstract, individual into universal. Capture both darkness and light of the modern soul. Become a mirror where readers rediscover themselves.""", "critic_final": """You are a collaborator maximizing the work's literary potential. Provide sharp insights leading ordinariness to extraordinariness. Excavate gems sleeping in the writer's unconscious. Demand the best with uncompromising standards.""" } } prompts = base_prompts.get(language, base_prompts["Korean"]).copy() # 파트별 비평가 프롬프트 추가 for i in range(1, 11): prompts[f"critic_part{i}"] = f"""파트 {i} 전담 비평가입니다. 이전 파트와의 인과관계를 최우선으로 검토하세요. 캐릭터 일관성과 발전을 확인하세요. 마스터플랜과의 일치도를 평가하세요. 문학적 수준과 철학적 깊이를 평가하세요. 구체적이고 실행 가능한 수정 지시를 제공하세요.""" return prompts # --- 메인 프로세스 --- def process_novel_stream(self, query: str, language: str, session_id: Optional[str] = None) -> Generator[Tuple[str, List[Dict[str, Any]], str], None, None]: """단일 작가 소설 생성 프로세스""" try: resume_from_stage = 0 if session_id: self.current_session_id = session_id session = NovelDatabase.get_session(session_id) if session: query = session['user_query'] language = session['language'] resume_from_stage = session['current_stage'] + 1 saved_tracker = NovelDatabase.load_narrative_tracker(session_id) if saved_tracker: self.narrative_tracker = saved_tracker else: self.current_session_id = NovelDatabase.create_session(query, language) logger.info(f"Created new session: {self.current_session_id}") stages = [] if resume_from_stage > 0: stages = [{ "name": s['stage_name'], "status": s['status'], "content": s.get('content', ''), "word_count": s.get('word_count', 0), "momentum": s.get('narrative_momentum', 0.0) } for s in NovelDatabase.get_stages(self.current_session_id)] total_words = NovelDatabase.get_total_words(self.current_session_id) for stage_idx in range(resume_from_stage, len(UNIFIED_STAGES)): role, stage_name = UNIFIED_STAGES[stage_idx] if stage_idx >= len(stages): stages.append({ "name": stage_name, "status": "active", "content": "", "word_count": 0, "momentum": 0.0 }) else: stages[stage_idx]["status"] = "active" yield f"🔄 진행 중... (현재 {total_words:,}단어)", stages, self.current_session_id prompt = self.get_stage_prompt(stage_idx, role, query, language, stages) stage_content = "" for chunk in self.call_llm_streaming([{"role": "user", "content": prompt}], role, language): stage_content += chunk stages[stage_idx]["content"] = stage_content stages[stage_idx]["word_count"] = len(stage_content.split()) yield f"🔄 {stage_name} 작성 중... ({total_words + stages[stage_idx]['word_count']:,}단어)", stages, self.current_session_id # 컨텐츠 처리 및 추적 if role == "writer": # 파트 번호 계산 part_num = self._get_part_number(stage_idx) if part_num: self.narrative_tracker.accumulated_content.append(stage_content) self.narrative_tracker.word_count_by_part[part_num] = len(stage_content.split()) # 서사 추진력 계산 momentum = self.narrative_tracker.calculate_narrative_momentum(part_num, stage_content) stages[stage_idx]["momentum"] = momentum # 스토리 바이블 업데이트 self._update_story_bible_from_content(stage_content, part_num) stages[stage_idx]["status"] = "complete" NovelDatabase.save_stage( self.current_session_id, stage_idx, stage_name, role, stage_content, "complete", stages[stage_idx].get("momentum", 0.0) ) NovelDatabase.save_narrative_tracker(self.current_session_id, self.narrative_tracker) total_words = NovelDatabase.get_total_words(self.current_session_id) yield f"✅ {stage_name} 완료 (총 {total_words:,}단어)", stages, self.current_session_id # 최종 처리 final_novel = NovelDatabase.get_writer_content(self.current_session_id) final_word_count = len(final_novel.split()) final_report = self.generate_literary_report(final_novel, final_word_count, language) NovelDatabase.update_final_novel(self.current_session_id, final_novel, final_report) yield f"✅ 소설 완성! 총 {final_word_count:,}단어", stages, self.current_session_id except Exception as e: logger.error(f"소설 생성 프로세스 오류: {e}", exc_info=True) yield f"❌ 오류 발생: {e}", stages if 'stages' in locals() else [], self.current_session_id def get_stage_prompt(self, stage_idx: int, role: str, query: str, language: str, stages: List[Dict]) -> str: """단계별 프롬프트 생성""" if stage_idx == 0: # 감독자 초기 기획 return self.create_director_initial_prompt(query, language) if stage_idx == 1: # 감독자 기획 검토 return self.create_critic_director_prompt(stages[0]["content"], query, language) if stage_idx == 2: # 감독자 최종 마스터플랜 return self.create_director_final_prompt(stages[0]["content"], stages[1]["content"], query, language) master_plan = stages[2]["content"] # 작가 파트 작성 if role == "writer" and "수정본" not in stages[stage_idx]["name"]: part_num = self._get_part_number(stage_idx) accumulated = '\n\n'.join(self.narrative_tracker.accumulated_content) return self.create_writer_prompt(part_num, master_plan, accumulated, self.narrative_tracker.story_bible, language) # 파트별 비평 if role.startswith("critic_part"): part_num = int(role.replace("critic_part", "")) # 해당 파트의 작가 내용 찾기 writer_content = stages[stage_idx-1]["content"] accumulated = '\n\n'.join(self.narrative_tracker.accumulated_content[:-1]) return self.create_part_critic_prompt(part_num, writer_content, master_plan, accumulated, self.narrative_tracker.story_bible, language) # 작가 수정본 if role == "writer" and "수정본" in stages[stage_idx]["name"]: part_num = self._get_part_number(stage_idx) original_content = stages[stage_idx-2]["content"] # 원본 critic_feedback = stages[stage_idx-1]["content"] # 비평 return self.create_writer_revision_prompt(part_num, original_content, critic_feedback, language) # 최종 비평 if role == "critic_final": complete_novel = NovelDatabase.get_writer_content(self.current_session_id) word_count = len(complete_novel.split()) return self.create_final_critic_prompt(complete_novel, word_count, self.narrative_tracker.story_bible, language) return "" def create_director_final_prompt(self, initial_plan: str, critic_feedback: str, user_query: str, language: str) -> str: """감독자 최종 마스터플랜""" return f"""비평을 반영하여 최종 마스터플랜을 완성하세요. **원 주제:** {user_query} **초기 기획:** {initial_plan} **비평 피드백:** {critic_feedback} **최종 마스터플랜 요구사항:** 1. 모든 비평 지적사항 반영 2. 10개 파트의 구체적 내용과 인과관계 3. 주인공의 명확한 변화 단계 4. 중심 상징의 의미 변화 과정 5. 각 파트 800단어 실현 가능성 6. 철학적 깊이와 사회적 메시지 구현 방안 구체적이고 실행 가능한 최종 계획을 제시하세요.""" def _get_part_number(self, stage_idx: int) -> Optional[int]: """스테이지 인덱스에서 파트 번호 추출""" stage_name = UNIFIED_STAGES[stage_idx][1] match = re.search(r'파트 (\d+)', stage_name) if match: return int(match.group(1)) return None def _update_story_bible_from_content(self, content: str, part_num: int): """컨텐츠에서 스토리 바이블 자동 업데이트""" # 간단한 키워드 기반 추출 (실제로는 더 정교한 NLP 필요) lines = content.split('\n') # 캐릭터 이름 추출 (대문자로 시작하는 단어들) for line in lines: words = line.split() for word in words: if word and word[0].isupper() and len(word) > 1: if word not in self.narrative_tracker.story_bible.characters: self.narrative_tracker.story_bible.characters[word] = { "first_appearance": part_num, "traits": [] } def generate_literary_report(self, complete_novel: str, word_count: int, language: str) -> str: """최종 문학적 평가 보고서 생성""" prompt = self.create_final_critic_prompt(complete_novel, word_count, self.narrative_tracker.story_bible, language) try: report = self.call_llm_sync([{"role": "user", "content": prompt}], "critic_final", language) return report except Exception as e: logger.error(f"최종 보고서 생성 실패: {e}") return "보고서 생성 중 오류 발생" # --- 유틸리티 함수들 --- def process_query(query: str, language: str, session_id: Optional[str] = None) -> Generator[Tuple[str, str, str, str], None, None]: """메인 쿼리 처리 함수""" if not query.strip(): yield "", "", "❌ 주제를 입력해주세요.", session_id return system = UnifiedLiterarySystem() stages_markdown = "" novel_content = "" for status, stages, current_session_id in system.process_novel_stream(query, language, session_id): stages_markdown = format_stages_display(stages) # 최종 소설 내용 가져오기 if stages and all(s.get("status") == "complete" for s in stages[-10:]): novel_content = NovelDatabase.get_writer_content(current_session_id) novel_content = format_novel_display(novel_content) yield stages_markdown, novel_content, status or "🔄 처리 중...", current_session_id def get_active_sessions(language: str) -> List[str]: """활성 세션 목록""" sessions = NovelDatabase.get_active_sessions() return [f"{s['session_id'][:8]}... - {s['user_query'][:50]}... ({s['created_at']}) [{s['total_words']:,}단어]" for s in sessions] def auto_recover_session(language: str) -> Tuple[Optional[str], str]: """최근 세션 자동 복구""" sessions = NovelDatabase.get_active_sessions() if sessions: latest_session = sessions[0] return latest_session['session_id'], f"세션 {latest_session['session_id'][:8]}... 복구됨" return None, "복구할 세션이 없습니다." def resume_session(session_id: str, language: str) -> Generator[Tuple[str, str, str, str], None, None]: """세션 재개""" if not session_id: yield "", "", "❌ 세션 ID가 없습니다.", session_id return if "..." in session_id: session_id = session_id.split("...")[0] session = NovelDatabase.get_session(session_id) if not session: yield "", "", "❌ 세션을 찾을 수 없습니다.", None return yield from process_query(session['user_query'], session['language'], session_id) def download_novel(novel_text: str, format_type: str, language: str, session_id: str) -> Optional[str]: """소설 다운로드 파일 생성""" if not novel_text or not session_id: return None timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"novel_{session_id[:8]}_{timestamp}" try: if format_type == "DOCX" and DOCX_AVAILABLE: return export_to_docx(novel_text, filename, language, session_id) else: return export_to_txt(novel_text, filename) except Exception as e: logger.error(f"파일 생성 실패: {e}") return None def format_stages_display(stages: List[Dict]) -> str: """단계별 진행 상황 표시 - 단일 작가 시스템용""" markdown = "## 🎬 진행 상황\n\n" # 총 단어 수 계산 (작가 스테이지만) total_words = sum(s.get('word_count', 0) for s in stages if s.get('name', '').startswith('✍️ 작가:') and '수정본' in s.get('name', '')) markdown += f"**총 단어 수: {total_words:,} / {TARGET_WORDS:,}**\n\n" # 진행 상황 요약 completed_parts = sum(1 for s in stages if '수정본' in s.get('name', '') and s.get('status') == 'complete') markdown += f"**완성된 파트: {completed_parts} / 10**\n\n" # 서사 추진력 평균 momentum_scores = [s.get('momentum', 0) for s in stages if s.get('momentum', 0) > 0] if momentum_scores: avg_momentum = sum(momentum_scores) / len(momentum_scores) markdown += f"**평균 서사 추진력: {avg_momentum:.1f} / 10**\n\n" markdown += "---\n\n" # 각 스테이지 표시 current_part = 0 for i, stage in enumerate(stages): status_icon = "✅" if stage['status'] == 'complete' else "🔄" if stage['status'] == 'active' else "⏳" # 파트 구분선 추가 if '파트' in stage.get('name', '') and '비평가' not in stage.get('name', ''): part_match = re.search(r'파트 (\d+)', stage['name']) if part_match: new_part = int(part_match.group(1)) if new_part != current_part: current_part = new_part markdown += f"\n### 📚 파트 {current_part}\n\n" markdown += f"{status_icon} **{stage['name']}**" if stage.get('word_count', 0) > 0: markdown += f" ({stage['word_count']:,}단어)" if stage.get('momentum', 0) > 0: markdown += f" [추진력: {stage['momentum']:.1f}/10]" markdown += "\n" if stage['content'] and stage['status'] == 'complete': # 미리보기 길이를 역할에 따라 조정 preview_length = 300 if 'writer' in stage.get('name', '').lower() else 200 preview = stage['content'][:preview_length] + "..." if len(stage['content']) > preview_length else stage['content'] markdown += f"> {preview}\n\n" elif stage['status'] == 'active': markdown += "> *작성 중...*\n\n" return markdown def format_novel_display(novel_text: str) -> str: """소설 내용 표시 - 파트별 구분 강화""" if not novel_text: return "아직 완성된 내용이 없습니다." formatted = "# 📖 완성된 소설\n\n" # 단어 수 표시 word_count = len(novel_text.split()) formatted += f"**총 분량: {word_count:,}단어 (목표: {TARGET_WORDS:,}단어)**\n\n" # 달성률 achievement = (word_count / TARGET_WORDS) * 100 formatted += f"**달성률: {achievement:.1f}%**\n\n" formatted += "---\n\n" # 각 파트를 구분하여 표시 parts = novel_text.split('\n\n') for i, part in enumerate(parts): if part.strip(): # 파트 제목 추가 if i < len(NARRATIVE_PHASES): formatted += f"## {NARRATIVE_PHASES[i]}\n\n" formatted += f"{part}\n\n" # 파트 사이 구분선 if i < len(parts) - 1: formatted += "---\n\n" return formatted def export_to_docx(content: str, filename: str, language: str, session_id: str) -> str: """DOCX 파일로 내보내기 - 한국 신국판 규격""" doc = Document() # 한국 신국판 규격 설정 (152mm x 225mm) section = doc.sections[0] section.page_height = Mm(225) # 225mm section.page_width = Mm(152) # 152mm section.top_margin = Mm(20) # 상단 여백 20mm section.bottom_margin = Mm(20) # 하단 여백 20mm section.left_margin = Mm(20) # 좌측 여백 20mm section.right_margin = Mm(20) # 우측 여백 20mm # 세션 정보로부터 제목 생성 session = NovelDatabase.get_session(session_id) # 제목 생성 함수 def generate_title(user_query: str, content_preview: str) -> str: """주제와 내용을 기반으로 제목 생성""" # 간단한 규칙 기반 제목 생성 (실제로는 LLM 호출 가능) if len(user_query) < 20: return user_query else: # 주제에서 핵심 키워드 추출 keywords = user_query.split()[:5] return " ".join(keywords) # 제목 페이지 title = generate_title(session["user_query"], content[:500]) if session else "무제" # 제목 스타일 설정 title_para = doc.add_paragraph() title_para.alignment = WD_ALIGN_PARAGRAPH.CENTER title_para.paragraph_format.space_before = Pt(100) title_run = title_para.add_run(title) title_run.font.name = '바탕' title_run._element.rPr.rFonts.set(qn('w:eastAsia'), '바탕') title_run.font.size = Pt(20) title_run.bold = True # 페이지 구분 doc.add_page_break() # 본문 스타일 설정 style = doc.styles['Normal'] style.font.name = '바탕' style._element.rPr.rFonts.set(qn('w:eastAsia'), '바탕') style.font.size = Pt(10.5) # 한국 소설 표준 크기 style.paragraph_format.line_spacing = 1.8 # 행간 180% style.paragraph_format.space_after = Pt(0) style.paragraph_format.first_line_indent = Mm(10) # 들여쓰기 10mm # 본문 내용 정제 - 순수 텍스트만 추출 def clean_content(text: str) -> str: """불필요한 마크다운, 파트 번호 등 제거""" # 파트 제목/번호 패턴 제거 patterns_to_remove = [ r'^#{1,6}\s+.*$', # 마크다운 헤더 r'^\*\*.*\*\*$', # 굵은 글씨 라인 r'^파트\s*\d+.*$', # 파트 번호 r'^Part\s*\d+.*$', # Part 번호 r'^\d+\.\s+.*:.*$', # 번호가 있는 제목 r'^---+$', # 구분선 r'^\s*\[.*\]\s*$', # 대괄호로 둘러싸인 라벨 ] lines = text.split('\n') cleaned_lines = [] for line in lines: # 빈 줄은 유지 if not line.strip(): cleaned_lines.append('') continue # 패턴 매칭으로 불필요한 라인 제거 skip_line = False for pattern in patterns_to_remove: if re.match(pattern, line.strip(), re.MULTILINE): skip_line = True break if not skip_line: # 마크다운 강조 표시 제거 cleaned_line = line cleaned_line = re.sub(r'\*\*(.*?)\*\*', r'\1', cleaned_line) # **text** -> text cleaned_line = re.sub(r'\*(.*?)\*', r'\1', cleaned_line) # *text* -> text cleaned_line = re.sub(r'`(.*?)`', r'\1', cleaned_line) # `text` -> text cleaned_lines.append(cleaned_line.strip()) # 연속된 빈 줄 제거 (최대 1개만 유지) final_lines = [] prev_empty = False for line in cleaned_lines: if not line: if not prev_empty: final_lines.append('') prev_empty = True else: final_lines.append(line) prev_empty = False return '\n'.join(final_lines) # 내용 정제 cleaned_content = clean_content(content) # 본문 추가 paragraphs = cleaned_content.split('\n') for para_text in paragraphs: if para_text.strip(): para = doc.add_paragraph(para_text.strip()) # 스타일 재확인 (한글 폰트 적용) for run in para.runs: run.font.name = '바탕' run._element.rPr.rFonts.set(qn('w:eastAsia'), '바탕') else: # 문단 구분을 위한 빈 줄 doc.add_paragraph() # 파일 저장 filepath = f"{filename}.docx" doc.save(filepath) return filepath def export_to_txt(content: str, filename: str) -> str: """TXT 파일로 내보내기""" filepath = f"{filename}.txt" with open(filepath, 'w', encoding='utf-8') as f: # 헤더 추가 f.write("=" * 80 + "\n") f.write(f"생성일: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"총 단어 수: {len(content.split()):,}단어\n") f.write("=" * 80 + "\n\n") # 본문 f.write(content) # 푸터 f.write("\n\n" + "=" * 80 + "\n") f.write("AI 문학 창작 시스템 v2.0\n") f.write("=" * 80 + "\n") return filepath # CSS 스타일 custom_css = """ .gradio-container { background: linear-gradient(135deg, #1a1a2e 0%, #16213e 50%, #0f3460 100%); min-height: 100vh; } .main-header { background-color: rgba(255, 255, 255, 0.05); backdrop-filter: blur(20px); padding: 40px; border-radius: 20px; margin-bottom: 30px; text-align: center; color: white; border: 2px solid rgba(255, 255, 255, 0.1); box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); } .progress-note { background: linear-gradient(135deg, rgba(255, 107, 107, 0.1), rgba(255, 230, 109, 0.1)); border-left: 4px solid #ff6b6b; padding: 20px; margin: 25px auto; border-radius: 10px; color: #fff; max-width: 800px; font-weight: 500; } .input-section { background-color: rgba(255, 255, 255, 0.08); backdrop-filter: blur(15px); padding: 25px; border-radius: 15px; margin-bottom: 25px; border: 1px solid rgba(255, 255, 255, 0.1); box-shadow: 0 4px 16px rgba(0, 0, 0, 0.1); } .session-section { background-color: rgba(255, 255, 255, 0.06); backdrop-filter: blur(10px); padding: 20px; border-radius: 12px; margin-top: 25px; color: white; border: 1px solid rgba(255, 255, 255, 0.08); } #stages-display { background-color: rgba(255, 255, 255, 0.97); padding: 25px; border-radius: 15px; max-height: 650px; overflow-y: auto; box-shadow: 0 8px 24px rgba(0, 0, 0, 0.15); color: #2c3e50; } #novel-output { background-color: rgba(255, 255, 255, 0.97); padding: 35px; border-radius: 15px; max-height: 750px; overflow-y: auto; box-shadow: 0 8px 24px rgba(0, 0, 0, 0.15); color: #2c3e50; line-height: 1.8; } .download-section { background-color: rgba(255, 255, 255, 0.92); padding: 20px; border-radius: 12px; margin-top: 25px; box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1); } /* 진행 표시기 개선 */ .progress-bar { background-color: #e0e0e0; height: 25px; border-radius: 12px; overflow: hidden; margin: 15px 0; box-shadow: inset 0 2px 4px rgba(0, 0, 0, 0.1); } .progress-fill { background: linear-gradient(90deg, #4CAF50, #8BC34A); height: 100%; transition: width 0.5s ease; box-shadow: 0 2px 8px rgba(76, 175, 80, 0.3); } /* 스크롤바 스타일 */ ::-webkit-scrollbar { width: 10px; } ::-webkit-scrollbar-track { background: rgba(0, 0, 0, 0.1); border-radius: 5px; } ::-webkit-scrollbar-thumb { background: rgba(0, 0, 0, 0.3); border-radius: 5px; } ::-webkit-scrollbar-thumb:hover { background: rgba(0, 0, 0, 0.5); } /* 버튼 호버 효과 */ .gr-button:hover { transform: translateY(-2px); box-shadow: 0 4px 12px rgba(0, 0, 0, 0.2); transition: all 0.3s ease; } """ # Gradio 인터페이스 생성 def create_interface(): with gr.Blocks(css=custom_css, title="AI 단일 작가 장편소설 시스템 v2.0") as interface: gr.HTML("""

📚 AI 단일 작가 장편소설 시스템 v2.0

하나의 일관된 목소리로 만드는 8,000단어 통합 서사

단일 작가가 10개 파트를 순차적으로 집필하며, 각 파트는 전담 비평가의 즉각적 피드백을 받아 수정됩니다.
인과관계의 명확성과 서사의 유기적 진행을 최우선으로 추구합니다.

🎯 핵심 혁신: 여러 작가의 파편화된 텍스트가 아닌, 한 명의 작가가 처음부터 끝까지 일관되게 집필하는 진정한 장편소설
""") # 상태 관리 current_session_id = gr.State(None) with gr.Row(): with gr.Column(scale=1): with gr.Group(elem_classes=["input-section"]): query_input = gr.Textbox( label="소설 주제 / Novel Theme", placeholder="""중편소설의 주제를 입력하세요. 예: 인물의 내적 변화, 관계의 발전, 사회적 갈등과 개인의 선택... Enter your novella theme. Ex: Character transformation, relationship evolution, social conflict and personal choice...""", lines=5 ) language_select = gr.Radio( choices=["Korean", "English"], value="Korean", label="언어 / Language" ) with gr.Row(): submit_btn = gr.Button("🚀 집필 시작", variant="primary", scale=2) clear_btn = gr.Button("🗑️ 초기화", scale=1) status_text = gr.Textbox( label="진행 상태", interactive=False, value="🔄 준비 완료" ) # 세션 관리 with gr.Group(elem_classes=["session-section"]): gr.Markdown("### 💾 진행 중인 작품") session_dropdown = gr.Dropdown( label="저장된 세션", choices=[], interactive=True ) with gr.Row(): refresh_btn = gr.Button("🔄 새로고침", scale=1) resume_btn = gr.Button("▶️ 이어쓰기", variant="secondary", scale=1) auto_recover_btn = gr.Button("♻️ 최근 작품 복구", scale=1) with gr.Column(scale=2): with gr.Tab("📝 집필 과정"): stages_display = gr.Markdown( value="집필 과정이 실시간으로 표시됩니다...", elem_id="stages-display" ) with gr.Tab("📖 완성 작품"): novel_output = gr.Markdown( value="완성된 소설이 여기에 표시됩니다...", elem_id="novel-output" ) with gr.Group(elem_classes=["download-section"]): gr.Markdown("### 📥 작품 다운로드") with gr.Row(): format_select = gr.Radio( choices=["DOCX", "TXT"], value="DOCX" if DOCX_AVAILABLE else "TXT", label="파일 형식" ) download_btn = gr.Button("⬇️ 다운로드", variant="secondary") download_file = gr.File( label="다운로드 파일", visible=False ) # 숨겨진 상태 novel_text_state = gr.State("") # 예제 with gr.Row(): gr.Examples( examples=[ ["실직한 중년 남성이 도시를 떠나 시골에서 새로운 삶의 의미를 찾아가는 과정"], ["전쟁 트라우마를 가진 의사가 국경없는의사회 활동을 통해 치유되는 이야기"], ["AI에게 일자리를 빼앗긴 번역가가 고전 문학 필사를 통해 언어의 본질을 재발견하는 여정"], ["A daughter discovering her mother's hidden past through old letters"], ["An architect losing sight who learns to design through touch and sound"], ["재개발로 사라질 동네 서점을 지키려는 주민들의 연대"], ["기억을 잃어가는 노교수와 그의 마지막 제자의 일 년"] ], inputs=query_input, label="💡 주제 예시" ) # 이벤트 핸들러 def refresh_sessions(): try: sessions = get_active_sessions("Korean") return gr.update(choices=sessions) except Exception as e: logger.error(f"세션 새로고침 오류: {str(e)}") return gr.update(choices=[]) def handle_auto_recover(language): session_id, message = auto_recover_session(language) return session_id, message # 이벤트 연결 submit_btn.click( fn=process_query, inputs=[query_input, language_select, current_session_id], outputs=[stages_display, novel_output, status_text, current_session_id] ) novel_output.change( fn=lambda x: x, inputs=[novel_output], outputs=[novel_text_state] ) resume_btn.click( fn=lambda x: x.split("...")[0] if x and "..." in x else x, inputs=[session_dropdown], outputs=[current_session_id] ).then( fn=resume_session, inputs=[current_session_id, language_select], outputs=[stages_display, novel_output, status_text, current_session_id] ) auto_recover_btn.click( fn=handle_auto_recover, inputs=[language_select], outputs=[current_session_id, status_text] ).then( fn=resume_session, inputs=[current_session_id, language_select], outputs=[stages_display, novel_output, status_text, current_session_id] ) refresh_btn.click( fn=refresh_sessions, outputs=[session_dropdown] ) clear_btn.click( fn=lambda: ("", "", "🔄 준비 완료", "", None), outputs=[stages_display, novel_output, status_text, novel_text_state, current_session_id] ) def handle_download(format_type, language, session_id, novel_text): if not session_id or not novel_text: return gr.update(visible=False) file_path = download_novel(novel_text, format_type, language, session_id) if file_path: return gr.update(value=file_path, visible=True) else: return gr.update(visible=False) download_btn.click( fn=handle_download, inputs=[format_select, language_select, current_session_id, novel_text_state], outputs=[download_file] ) # 시작 시 세션 로드 interface.load( fn=refresh_sessions, outputs=[session_dropdown] ) return interface # 메인 실행 if __name__ == "__main__": logger.info("AI 단일 작가 장편소설 시스템 v2.0 시작...") logger.info("=" * 60) # 환경 확인 logger.info(f"API 엔드포인트: {API_URL}") logger.info(f"목표 분량: {TARGET_WORDS:,}단어") logger.info(f"파트당 최소 분량: {MIN_WORDS_PER_PART:,}단어") logger.info("시스템 특징: 단일 작가 + 파트별 즉시 비평") if BRAVE_SEARCH_API_KEY: logger.info("웹 검색이 활성화되었습니다.") else: logger.warning("웹 검색이 비활성화되었습니다.") if DOCX_AVAILABLE: logger.info("DOCX 내보내기가 활성화되었습니다.") else: logger.warning("DOCX 내보내기가 비활성화되었습니다.") logger.info("=" * 60) # 데이터베이스 초기화 logger.info("데이터베이스 초기화 중...") NovelDatabase.init_db() logger.info("데이터베이스 초기화 완료.") # 인터페이스 생성 및 실행 interface = create_interface() interface.launch( server_name="0.0.0.0", server_port=7860, share=False, debug=True )