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 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_v4.db" # --- 환경 변수 검증 --- 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() # 문학적 단계 구성 (내면 서술과 사회적 통찰 중심) LITERARY_STAGES = [ ("director", "🎬 감독자: 사회적 맥락과 인물 심리 기획"), ("critic", "📝 비평가: 사회 비판적 깊이와 상징성 검토"), ("director", "🎬 감독자: 수정된 마스터플랜"), ] + [ (f"writer{i}", f"✍️ 작가 {i}: 초안") for i in range(1, 11) ] + [ ("critic", "📝 비평가: 중간 검토 (내적 일관성과 주제 심화)"), ] + [ (f"writer{i}", f"✍️ 작가 {i}: 수정본") for i in range(1, 11) ] + [ ("critic", f"📝 비평가: 최종 검토 및 문학적 평가"), ] # --- 데이터 클래스 --- @dataclass class CharacterPsychology: """인물의 심리적 상태와 내면""" name: str age: int social_class: str # 계급적 위치 occupation: str inner_conflict: str # 내적 갈등 worldview: str # 세계관 desires: List[str] # 욕망들 fears: List[str] # 두려움들 coping_mechanisms: List[str] # 방어기제 relationships: Dict[str, str] # 타인과의 관계 @dataclass class SymbolicElement: """상징적 요소""" symbol: str meaning: str appearances: List[int] # 등장하는 장들 evolution: str # 상징의 의미 변화 @dataclass class SocialContext: """사회적 맥락""" economic_system: str class_structure: str power_dynamics: str social_issues: List[str] cultural_atmosphere: str # --- 핵심 로직 클래스 --- class LiteraryConsistencyTracker: """문학적 일관성 추적 시스템""" def __init__(self): self.characters: Dict[str, CharacterPsychology] = {} self.symbols: Dict[str, SymbolicElement] = {} self.social_context: Optional[SocialContext] = None self.themes: List[str] = [] self.narrative_voice: str = "" # 서술 시점과 문체 self.tone: str = "" # 전체적인 톤 def register_character(self, character: CharacterPsychology): """인물 등록""" self.characters[character.name] = character logger.info(f"Character registered: {character.name}, Class: {character.social_class}") def register_symbol(self, symbol: SymbolicElement): """상징 등록""" self.symbols[symbol.symbol] = symbol logger.info(f"Symbol registered: {symbol.symbol} = {symbol.meaning}") def check_thematic_consistency(self, content: str, chapter: int) -> List[str]: """주제적 일관성 검사""" issues = [] # 사회비판적 요소가 유지되는지 if self.social_context and not any(issue.lower() in content.lower() for issue in self.social_context.social_issues): issues.append("사회적 맥락이 약화되었습니다. 계급, 불평등 등의 주제를 유지하세요.") # 내면 서술이 충분한지 introspective_keywords = ['생각했다', '느꼈다', '기억', '의식', '마음', 'thought', 'felt', 'remembered', 'consciousness'] if not any(keyword in content for keyword in introspective_keywords): issues.append("내면 서술이 부족합니다. 인물의 심리를 더 깊이 탐구하세요.") return issues 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, social_context TEXT, narrative_style 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', literary_score 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 characters ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, name TEXT NOT NULL, age INTEGER, social_class TEXT, occupation TEXT, inner_conflict TEXT, worldview TEXT, created_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (session_id) REFERENCES sessions(session_id), UNIQUE(session_id, name) ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS symbols ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, symbol TEXT NOT NULL, meaning 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', literary_score: 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, literary_score) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(session_id, stage_number) DO UPDATE SET content=?, word_count=?, status=?, stage_name=?, literary_score=?, updated_at=datetime('now') ''', (session_id, stage_number, stage_name, role, content, word_count, status, literary_score, content, word_count, status, stage_name, literary_score)) cursor.execute( "UPDATE sessions SET updated_at = datetime('now'), current_stage = ? WHERE session_id = ?", (stage_number, session_id) ) conn.commit() @staticmethod def get_writer_content(session_id: str) -> str: """작가 콘텐츠 가져오기""" with NovelDatabase.get_db() as conn: all_content = [] for writer_num in range(1, 11): row = conn.cursor().execute( "SELECT content FROM stages WHERE session_id = ? AND role = ? AND stage_name LIKE '%수정본%' ORDER BY stage_number DESC LIMIT 1", (session_id, f'writer{writer_num}') ).fetchone() if row and row['content']: content = row['content'].strip() all_content.append(content) return '\n\n'.join(all_content) @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 FROM sessions WHERE status = 'active' ORDER BY updated_at DESC LIMIT 10" ).fetchall() return [dict(row) for row in rows] 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 LiteraryNovelSystem: """문학적 소설 생성 시스템""" def __init__(self): self.token = FRIENDLI_TOKEN self.api_url = API_URL self.model_id = MODEL_ID self.consistency_tracker = LiteraryConsistencyTracker() 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 create_director_initial_prompt(self, user_query: str, language: str) -> str: """감독자 초기 기획 프롬프트 (사회비판적 리얼리즘)""" search_results_str = "" if self.web_search.enabled: # 사회적 맥락 검색 queries = [f"{user_query} 사회 문제", f"{user_query} 계급 갈등", f"{user_query} social inequality"] for q in queries[:1]: results = self.web_search.search(q, count=2, language=language) if results: search_results_str += self.web_search.extract_relevant_info(results) + "\n" lang_prompts = { "Korean": f"""당신은 한국 현대 문학의 전통을 잇는 소설 기획자입니다. 한강, 김애란, 천명관, 정세랑 등 현대 한국 작가들의 문학적 성취를 참고하여, 사회비판적 리얼리즘과 내면 심리 탐구가 결합된 30페이지 중편소설을 기획하세요. **사용자 주제:** {user_query} **참고 자료:** {search_results_str if search_results_str else "N/A"} **기획 요구사항:** 1. **주제와 사회적 맥락** - 핵심 주제: 현대 한국 사회의 구조적 문제 (계급, 빈곤, 소외, 젠더, 세대 등) - 비판적 시각: 개인의 문제를 사회 구조와 연결 - 현실성: 2020년대 한국의 구체적 현실 반영 2. **서술 방식과 문체** - 시점: 1인칭 또는 제한적 3인칭 (내면 접근 가능) - 문체: 담담하지만 날카로운 관찰, 일상어와 문학적 표현의 균형 - 내면 서술: 의식의 흐름, 회상, 자기 성찰 적극 활용 3. **인물 설정** (2-4명의 핵심 인물) | 이름 | 나이 | 계급적 위치 | 직업 | 내적 갈등 | 욕망 | 두려움 | - 각 인물은 특정 계급과 사회적 위치를 대표 - 겉모습과 내면의 괴리 표현 - 구조적 억압 속에서의 개인적 선택 4. **상징과 은유** - 핵심 상징: (예: 개구리=소외된 자들의 목소리, 연못=계급 경계) - 반복되는 이미지나 모티프 - 일상적 사물에 담긴 사회적 의미 5. **플롯 구조** - 극적 사건보다 일상 속 미묘한 변화 중심 - 인물의 인식 변화가 곧 서사의 진행 - 열린 결말: 해결보다는 문제 제기와 성찰 **절대 금지사항:** - 선악 구분이 명확한 평면적 인물 - 급작스러운 사건이나 극적 반전 - 교훈적이거나 계몽적인 메시지 - 안이한 희망이나 화해 내면과 사회를 동시에 포착하는 섬세한 기획을 작성하세요.""", "English": f"""You are a literary director planning a 30-page novella in the tradition of contemporary social realism. Drawing from authors like George Saunders, Zadie Smith, and Sally Rooney, create a work that combines psychological depth with social critique. **User Theme:** {user_query} **Reference Material:** {search_results_str if search_results_str else "N/A"} **Planning Requirements:** 1. **Theme and Social Context** - Core theme: Structural problems in contemporary society (class, inequality, alienation) - Critical perspective: Connect individual struggles to social systems - Realism: Reflect specific contemporary realities 2. **Narrative Style** - POV: First person or limited third person (with access to interiority) - Style: Understated yet sharp observation, balance of vernacular and literary - Interior narration: Stream of consciousness, memory, self-reflection 3. **Character Design** (2-4 main characters) | Name | Age | Class Position | Occupation | Inner Conflict | Desires | Fears | - Each character represents specific social position - Gap between appearance and interior life - Individual choices within structural constraints 4. **Symbols and Metaphors** - Key symbols with social meaning - Recurring images or motifs - Everyday objects as social commentary 5. **Plot Structure** - Focus on subtle changes over dramatic events - Character perception shifts drive narrative - Open ending: Questions over resolutions **Absolutely Avoid:** - Clear-cut heroes and villains - Sudden dramatic events or twists - Didactic or preachy messages - Easy hope or reconciliation Create a nuanced plan that captures both interior life and social reality.""" } 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. **사회비판적 깊이** - 개인과 구조의 연결이 설득력 있는가? - 현실의 복잡성을 단순화하지 않았는가? - 진부한 사회 비판에 그치지 않았는가? 2. **문학적 완성도** - 내면 서술과 외적 현실의 균형 - 상징과 은유의 적절성 - 인물의 입체성과 신빙성 3. **현대성과 보편성** - 2020년대 한국의 특수성 반영 - 동시에 보편적 인간 조건 탐구 구체적 개선 방향을 제시하세요.""", "English": f"""You are a literary critic. Review the plan from these perspectives: **Original Theme:** {user_query} **Director's Plan:** {director_plan} **Review Criteria:** 1. **Social Critical Depth** - Is the connection between individual and structure convincing? - Does it avoid oversimplifying complex realities? - Does it go beyond clichéd social criticism? 2. **Literary Merit** - Balance of interiority and external reality - Effectiveness of symbols and metaphors - Character complexity and credibility 3. **Contemporary Relevance** - Reflects specific contemporary context - While exploring universal human conditions Provide specific improvements.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_writer_prompt(self, writer_number: int, director_plan: str, previous_content: str, user_query: str, language: str) -> str: """작가 프롬프트 (내면 서술 중심)""" lang_prompts = { "Korean": f"""당신은 작가 {writer_number}번입니다. 한국 현대 문학의 전통에 따라 작성하세요. **마스터플랜:** {director_plan} **이전 내용:** {previous_content[-2000:] if previous_content else "시작"} **작성 지침:** 1. **분량**: 1,300-1,500 단어 2. **서술 방식** - 인물의 내면 의식을 깊이 탐구 - 관찰과 성찰의 교차 - 현재와 과거의 자연스러운 오버랩 3. **문체** - 담담하면서도 날카로운 시선 - 일상어 속에 스며든 시적 표현 - 짧은 문장과 긴 문장의 리듬감 4. **내용 전개** - 큰 사건보다 작은 순간의 의미 - 대화는 최소화, 있어도 함축적 - 인물의 인식 변화가 곧 플롯 5. **사회적 맥락** - 개인의 일상에 스며든 구조적 억압 - 직접적 비판보다 간접적 드러냄 - 독자가 스스로 깨닫게 하는 서술 **반드시 포함할 요소:** - 인물의 내적 독백이나 의식의 흐름 - 구체적인 감각적 디테일 - 사회적 맥락을 암시하는 일상의 순간 - 상징이나 은유의 자연스러운 활용 깊이 있는 내면 탐구와 섬세한 사회 관찰을 보여주세요.""", "English": f"""You are Writer #{writer_number}. Write in the contemporary literary tradition. **Masterplan:** {director_plan} **Previous Content:** {previous_content[-2000:] if previous_content else "Beginning"} **Writing Guidelines:** 1. **Length**: 1,300-1,500 words 2. **Narrative Approach** - Deep exploration of character consciousness - Intersection of observation and reflection - Natural overlap of present and past 3. **Style** - Understated yet sharp perspective - Poetic expression within everyday language - Rhythm of short and long sentences 4. **Development** - Meaning in small moments over big events - Minimal, implicit dialogue - Character perception shifts as plot 5. **Social Context** - Structural oppression in daily life - Indirect rather than direct critique - Let readers discover meaning **Must Include:** - Interior monologue or stream of consciousness - Concrete sensory details - Everyday moments revealing social context - Natural use of symbols and metaphors Show deep interior exploration and subtle social observation.""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_critic_consistency_prompt(self, all_content: str, user_query: str, language: str) -> str: """비평가 중간 검토""" return f"""문학 비평가로서 현재까지의 작품을 검토하세요. **원 주제:** {user_query} **작품 내용 (최근 부분):** {all_content[-3000:]} **검토 항목:** 1. **내적 일관성** - 인물의 의식과 행동의 일치 - 서술 시점의 일관성 - 문체와 톤의 통일성 2. **주제 심화** - 초기 설정한 사회적 문제의 지속적 탐구 - 상징과 은유의 발전 - 깊이 있는 성찰의 누적 3. **문학적 완성도** - 진부함이나 상투성 회피 - 독창적 표현과 관찰 - 여운과 함축성 각 작가에게 구체적 개선 방향을 제시하세요.""" def create_writer_revision_prompt(self, writer_number: int, initial_content: str, critic_feedback: str, language: str) -> str: """작가 수정 프롬프트""" return f"""작가 {writer_number}번, 비평을 반영하여 수정하세요. **초안:** {initial_content} **비평:** {critic_feedback} **수정 방향:** 1. 내면 서술 강화 2. 사회적 맥락 심화 3. 문학적 표현 개선 4. 진부함 제거 수정본만 제시하세요.""" def create_critic_final_prompt(self, complete_novel: str, language: str) -> str: """최종 비평""" return f"""완성된 소설의 문학적 가치를 평가하세요. **작품 (일부):** {complete_novel[-3000:]} **평가 기준:** 1. **주제 의식 (30점)** - 사회 비판의 예리함 - 인간 조건에 대한 통찰 - 현대성과 보편성의 조화 2. **인물과 심리 (25점)** - 내면 묘사의 깊이 - 인물의 신빙성 - 복잡성과 모순의 포착 3. **문체와 기법 (25점)** - 문장의 완성도 - 상징과 은유의 효과 - 독창성과 참신함 4. **구조와 완결성 (20점)** - 전체 구성의 균형 - 여운과 개방성 - 독자에게 던지는 질문 **총점: /100점** 한국 현대 문학의 맥락에서 이 작품의 의의를 논하세요.""" # --- 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] payload = { "model": self.model_id, "messages": full_messages, "max_tokens": 10000, "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": """당신은 한강, 김애란, 천명관 등 한국 현대 작가들의 작품을 깊이 이해하는 문학 기획자입니다. 사회비판적 리얼리즘과 심리적 깊이를 결합한 작품을 기획하세요. 개인의 내면과 사회 구조의 상호작용을 섬세하게 포착하는 것이 핵심입니다.""", "critic": """당신은 현대 한국 문학을 깊이 이해하는 비평가입니다. 작품이 진부한 사회 비판이나 표면적 심리 묘사에 그치지 않고, 진정한 문학적 가치를 지니는지 엄격하게 평가하세요.""", "writer_base": """당신은 현대 한국 문학의 전통을 잇는 작가입니다. 내면 의식의 흐름과 날카로운 사회 관찰을 결합하여, 독자에게 깊은 여운을 남기는 문장을 쓰세요. '보여주기'보다 '의식하기'를, 사건보다 인식을 중시하세요.""" }, "English": { "director": """You are a literary planner deeply versed in contemporary social realist fiction. Plan works that combine social critique with psychological depth. The key is capturing the subtle interplay between individual consciousness and social structures.""", "critic": """You are a critic well-versed in contemporary literary fiction. Evaluate whether works go beyond superficial social commentary or psychology to achieve genuine literary value.""", "writer_base": """You are a writer in the contemporary literary tradition. Combine stream of consciousness with sharp social observation to create resonant prose. Prioritize consciousness over showing, perception over events.""" } } prompts = base_prompts.get(language, base_prompts["Korean"]).copy() # 특수 작가 프롬프트 if language == "Korean": prompts["writer1"] = prompts["writer_base"] + "\n특히 도입부에서 독자를 작품의 분위기로 서서히 끌어들이세요." prompts["writer5"] = prompts["writer_base"] + "\n중반부의 심리적 밀도를 높이고 갈등을 내면화하세요." prompts["writer10"] = prompts["writer_base"] + "\n열린 결말로 독자에게 질문을 던지세요." for i in range(2, 10): if f"writer{i}" not in prompts: prompts[f"writer{i}"] = prompts["writer_base"] 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 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', ''), "literary_score": s.get('literary_score', 0.0) } for s in NovelDatabase.get_stages(self.current_session_id)] for stage_idx in range(resume_from_stage, len(LITERARY_STAGES)): role, stage_name = LITERARY_STAGES[stage_idx] if stage_idx >= len(stages): stages.append({ "name": stage_name, "status": "active", "content": "", "literary_score": 0.0 }) else: stages[stage_idx]["status"] = "active" yield "", 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 yield "", stages, self.current_session_id # 문학적 점수 평가 literary_score = self.evaluate_literary_quality(stage_content, role) stages[stage_idx]["literary_score"] = literary_score stages[stage_idx]["status"] = "complete" NovelDatabase.save_stage( self.current_session_id, stage_idx, stage_name, role, stage_content, "complete", literary_score ) yield "", stages, self.current_session_id # 최종 소설 정리 final_novel = NovelDatabase.get_writer_content(self.current_session_id) final_report = self.generate_literary_report(final_novel, language) NovelDatabase.update_final_novel(self.current_session_id, final_novel, final_report) yield f"✅ 소설 완성! 총 {len(final_novel.split())}단어", 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_revision_prompt(stages[0]["content"], stages[1]["content"], query, language) master_plan = stages[2]["content"] if 3 <= stage_idx <= 12: # 작가 초안 writer_num = stage_idx - 2 previous_content = self.get_all_content(stages, stage_idx) return self.create_writer_prompt(writer_num, master_plan, previous_content, query, language) if stage_idx == 13: # 비평가 중간 검토 all_content = self.get_all_content(stages, stage_idx) return self.create_critic_consistency_prompt(all_content, query, language) if 14 <= stage_idx <= 23: # 작가 수정 writer_num = stage_idx - 13 initial_content = stages[2 + writer_num]["content"] feedback = stages[13]["content"] return self.create_writer_revision_prompt(writer_num, initial_content, feedback, language) if stage_idx == 24: # 최종 검토 complete_novel = self.get_all_writer_content(stages) return self.create_critic_final_prompt(complete_novel, language) return "" def create_director_revision_prompt(self, initial_plan: str, critic_feedback: str, user_query: str, language: str) -> str: """감독자 수정 프롬프트""" return f"""비평을 반영하여 기획을 수정하세요. **원 주제:** {user_query} **초기 기획:** {initial_plan} **비평:** {critic_feedback} **수정 방향:** 1. 사회비판적 깊이 강화 2. 인물의 내면 복잡성 증대 3. 상징과 은유 정교화 4. 진부함 제거 수정된 마스터플랜을 제시하세요.""" def get_all_content(self, stages: List[Dict], current_stage: int) -> str: """현재까지의 모든 내용""" contents = [] for i, s in enumerate(stages): if i < current_stage and s["content"] and "writer" in s.get("name", ""): contents.append(s["content"]) return "\n\n".join(contents) def get_all_writer_content(self, stages: List[Dict]) -> str: """모든 작가 최종본""" contents = [] for i, s in enumerate(stages): if 14 <= i <= 23 and s["content"]: contents.append(s["content"]) return "\n\n".join(contents) def evaluate_literary_quality(self, content: str, role: str) -> float: """문학적 품질 평가""" if not content or not role.startswith("writer"): return 0.0 score = 5.0 # 기본 점수 # 내면 서술 평가 introspective_patterns = ['생각했다', '느꼈다', '기억', '의식', '떠올렸다', '마음', '머릿속', '가슴', 'thought', 'felt', 'remembered'] introspection_count = sum(1 for pattern in introspective_patterns if pattern in content) score += min(2.0, introspection_count * 0.2) # 감각적 디테일 sensory_patterns = ['냄새', '소리', '빛', '그림자', '촉감', '맛', '온도', '색', 'smell', 'sound', 'light', 'shadow', 'touch', 'taste'] sensory_count = sum(1 for pattern in sensory_patterns if pattern in content) score += min(1.5, sensory_count * 0.15) # 사회적 맥락 social_patterns = ['임대', '계급', '빈곤', '격차', '차별', '소외', '불평등', 'rent', 'class', 'poverty', 'gap', 'discrimination'] social_count = sum(1 for pattern in social_patterns if pattern in content) score += min(1.5, social_count * 0.3) return min(10.0, score) def generate_literary_report(self, complete_novel: str, language: str) -> str: """최종 문학적 평가 보고서""" prompt = self.create_critic_final_prompt(complete_novel, language) try: report = self.call_llm_sync([{"role": "user", "content": prompt}], "critic", 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 = LiteraryNovelSystem() 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']})" for s in sessions] def auto_recover_session(language: str) -> Tuple[Optional[str], str]: """최근 세션 자동 복구""" latest_session = NovelDatabase.get_latest_active_session() if latest_session: 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" for i, stage in enumerate(stages): status_icon = "✅" if stage['status'] == 'complete' else "🔄" if stage['status'] == 'active' else "⏳" markdown += f"{status_icon} **{stage['name']}**" if stage.get('literary_score', 0) > 0: markdown += f" (문학성: {stage['literary_score']:.1f}/10)" markdown += "\n" if stage['content']: preview = stage['content'][:200] + "..." if len(stage['content']) > 200 else stage['content'] markdown += f"> {preview}\n\n" return markdown def format_novel_display(novel_text: str) -> str: """소설 내용 표시""" if not novel_text: return "아직 완성된 내용이 없습니다." formatted = "# 📖 완성된 소설\n\n" formatted += novel_text 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 = Inches(8.86) # 225mm section.page_width = Inches(5.98) # 152mm # 여백 설정 section.top_margin = Inches(0.79) # 20mm section.bottom_margin = Inches(0.79) section.left_margin = Inches(0.79) section.right_margin = Inches(0.79) # 세션 정보 session = NovelDatabase.get_session(session_id) # 제목 페이지 title_para = doc.add_paragraph() title_para.alignment = WD_ALIGN_PARAGRAPH.CENTER # 빈 줄 추가 for _ in range(8): doc.add_paragraph() if session: title_run = title_para.add_run(session["user_query"]) title_run.font.size = Pt(16) title_run.font.name = '바탕' # 페이지 나누기 doc.add_page_break() # 본문 스타일 설정 style = doc.styles['Normal'] style.font.name = '바탕' style.font.size = Pt(10.5) style.paragraph_format.line_spacing = 1.8 style.paragraph_format.alignment = WD_ALIGN_PARAGRAPH.JUSTIFY style.paragraph_format.first_line_indent = Inches(0.35) style.paragraph_format.space_after = Pt(3) # 본문 추가 paragraphs = content.split('\n\n') for para_text in paragraphs: if para_text.strip(): para = doc.add_paragraph(para_text.strip()) # 파일 저장 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(content) return filepath # CSS 스타일 (문학적 분위기) custom_css = """ .gradio-container { background: linear-gradient(135deg, #1a1a1a 0%, #2d2d2d 50%, #1a1a1a 100%); min-height: 100vh; } .main-header { background-color: rgba(255, 255, 255, 0.03); backdrop-filter: blur(10px); padding: 30px; border-radius: 12px; margin-bottom: 30px; text-align: center; color: white; border: 1px solid rgba(255, 255, 255, 0.1); } .literary-note { background-color: rgba(255, 255, 255, 0.05); border-left: 3px solid #888; padding: 15px; margin: 20px 0; border-radius: 8px; color: #ccc; font-style: italic; font-family: 'Georgia', serif; } .input-section { background-color: rgba(255, 255, 255, 0.05); backdrop-filter: blur(10px); padding: 20px; border-radius: 12px; margin-bottom: 20px; border: 1px solid rgba(255, 255, 255, 0.1); } .session-section { background-color: rgba(255, 255, 255, 0.05); backdrop-filter: blur(10px); padding: 15px; border-radius: 8px; margin-top: 20px; color: white; border: 1px solid rgba(255, 255, 255, 0.1); } #stages-display { background-color: rgba(255, 255, 255, 0.95); padding: 20px; border-radius: 12px; max-height: 600px; overflow-y: auto; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); } #novel-output { background-color: rgba(255, 255, 255, 0.98); padding: 40px; border-radius: 12px; max-height: 700px; overflow-y: auto; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); font-family: '바탕', 'Batang', 'Georgia', serif; line-height: 2; color: #333; } .download-section { background-color: rgba(255, 255, 255, 0.9); padding: 15px; border-radius: 8px; margin-top: 20px; box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1); } /* 문학적 스타일 */ #novel-output p { text-indent: 2em; margin-bottom: 1em; text-align: justify; } #novel-output h1 { color: #1a1a1a; font-weight: normal; text-align: center; margin: 2em 0; font-size: 1.8em; } /* 인용문이나 내적 독백 스타일 */ #novel-output blockquote { margin: 1.5em 2em; font-style: italic; color: #555; border-left: 3px solid #ccc; padding-left: 1em; } """ # Gradio 인터페이스 생성 def create_interface(): with gr.Blocks(css=custom_css, title="AI 문학 소설 생성 시스템") as interface: gr.HTML("""

📖 AI 문학 소설 생성 시스템

현대 한국 문학의 전통을 잇는 소설 창작

사회비판적 리얼리즘과 심리적 깊이를 결합한 30페이지 중편소설을 생성합니다.
개인의 내면과 사회 구조의 상호작용을 섬세하게 포착합니다.

"가장 개인적인 것이 가장 보편적인 것이다"
""") # 상태 관리 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="현대 사회의 문제나 인간의 조건에 대한 주제를 입력하세요...\nEnter themes about contemporary social issues or human condition...", lines=4 ) 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=[ ["도시 빈민의 주거권과 계급 갈등"], ["비정규직 청년의 불안과 소외"], ["돌봄 노동자의 보이지 않는 삶"], ["젠트리피케이션과 도시 난민"], ["The invisible labor of care workers"], ["Digital divide and social isolation"], ["세대 간 단절과 노인 빈곤"], ["환경 재난 시대의 불평등"] ], 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"Error refreshing sessions: {str(e)}") return gr.update(choices=[]) def handle_auto_recover(language): session_id, message = auto_recover_session(language) return session_id # 이벤트 연결 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, 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] ).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: 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 문학 소설 생성 시스템 시작...") logger.info("=" * 60) # 환경 확인 logger.info(f"API 엔드포인트: {API_URL}") 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 )