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 import random from huggingface_hub import HfApi, upload_file, hf_hub_download import replicate from PIL import Image import io as io_module import base64 import concurrent.futures from threading import Lock # --- Logging setup --- 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.") # --- Environment variables and constants --- FIREWORKS_API_KEY = os.getenv("FIREWORKS_API_KEY", "") REPLICATE_API_TOKEN = os.getenv("REPLICATE_API_TOKEN", "") BRAVE_SEARCH_API_KEY = os.getenv("BRAVE_SEARCH_API_KEY", "") API_URL = "https://api.fireworks.ai/inference/v1/chat/completions" MODEL_ID = "accounts/fireworks/models/qwen3-235b-a22b-instruct-2507" DB_PATH = "webtoon_sessions_v1.db" # Initialize Replicate client if token exists if REPLICATE_API_TOKEN: os.environ["REPLICATE_API_TOKEN"] = REPLICATE_API_TOKEN # Target settings for webtoon TARGET_EPISODES = 40 # 40화 완결 PANELS_PER_EPISODE = 30 # 각 화당 30개 패널 TARGET_PANELS = TARGET_EPISODES * PANELS_PER_EPISODE # 총 1200 패널 # Webtoon genres WEBTOON_GENRES = { "로맨스": "Romance", "로판": "Romance Fantasy", "판타지": "Fantasy", "현판": "Modern Fantasy", "무협": "Martial Arts", "스릴러": "Thriller", "일상": "Slice of Life", "개그": "Comedy", "스포츠": "Sports" } # Celebrity face references for character design CELEBRITY_FACES = { "male": [ "톰 크루즈", "브래드 피트", "레오나르도 디카프리오", "라이언 고슬링", "크리스 헴스워스", "로버트 다우니 주니어", "크리스 에반스", "톰 히들스턴", "베네딕트 컴버배치", "키아누 리브스", "이병헌", "공유", "박서준", "송중기" ], "female": [ "스칼렛 요한슨", "엠마 왓슨", "제니퍼 로렌스", "갤 가돗", "마고 로비", "엠마 스톤", "앤 해서웨이", "나탈리 포트만", "전지현", "송혜교", "김태리", "아이유", "수지", "한소희" ] } # --- Environment validation --- if not FIREWORKS_API_KEY: logger.error("FIREWORKS_API_KEY not set. Application will not work properly.") FIREWORKS_API_KEY = "dummy_token_for_testing" if not REPLICATE_API_TOKEN: logger.warning("REPLICATE_API_TOKEN not set. Image generation will be disabled.") # --- Global variables --- db_lock = threading.Lock() generated_images_cache = {} # Cache for generated images # --- Genre-specific prompts and elements --- GENRE_ELEMENTS = { "로맨스": { "key_elements": ["감정선", "오해와 화해", "달콤한 순간", "질투", "고백"], "visual_styles": ["소프트 톤", "파스텔", "꽃 배경", "빛망울 효과", "분홍빛 필터"], "panel_types": ["클로즈업 감정샷", "투샷", "손 클로즈업", "눈빛 교환", "백허그"], "typical_scenes": ["카페 데이트", "우산 씬", "불꽃놀이", "옥상 고백", "공항 이별"] }, "로판": { "key_elements": ["회귀/빙의", "드레스", "무도회", "마법", "신분 상승"], "visual_styles": ["화려한 의상", "유럽풍 배경", "반짝이 효과", "마법진", "성 배경"], "panel_types": ["전신샷", "드레스 디테일", "마법 이펙트", "회상씬", "충격 리액션"], "typical_scenes": ["무도회장", "정원 산책", "서재", "마법 수업", "알현실"] }, "판타지": { "key_elements": ["마법체계", "레벨업", "던전", "길드", "모험"], "visual_styles": ["다이나믹 액션", "이펙트 강조", "몬스터 디자인", "판타지 배경", "빛 효과"], "panel_types": ["액션씬", "풀샷 전투", "스킬 발동", "몬스터 등장", "파워업"], "typical_scenes": ["던전 입구", "보스전", "길드 회관", "수련장", "아이템 획득"] }, "현판": { "key_elements": ["게이트", "헌터", "각성", "현대 도시", "능력"], "visual_styles": ["도시 배경", "네온 효과", "현대적 액션", "특수 효과", "어반 판타지"], "panel_types": ["도시 전경", "능력 발현", "게이트 출현", "전투 액션", "일상 대비"], "typical_scenes": ["게이트 현장", "헌터 협회", "훈련장", "병원", "학교"] }, "무협": { "key_elements": ["무공", "문파", "강호", "복수", "의협"], "visual_styles": ["동양풍", "먹 효과", "기 표현", "중국풍 의상", "산수화 배경"], "panel_types": ["검술 동작", "경공술", "기공 수련", "대결 구도", "폭발 이펙트"], "typical_scenes": ["무림맹", "객잔", "절벽", "폭포 수련", "비무대회"] }, "스릴러": { "key_elements": ["서스펜스", "공포", "추격", "심리전", "반전"], "visual_styles": ["어두운 톤", "그림자 강조", "대비 효과", "불안한 구도", "붉은색 강조"], "panel_types": ["극단 클로즈업", "Dutch angle", "실루엣", "충격 컷", "공포 연출"], "typical_scenes": ["어두운 골목", "폐건물", "지하실", "추격씬", "대치 상황"] }, "일상": { "key_elements": ["일상", "공감", "소소한 재미", "관계", "성장"], "visual_styles": ["따뜻한 색감", "부드러운 선", "일상 배경", "캐주얼", "편안한 구도"], "panel_types": ["일상 컷", "리액션", "대화씬", "배경 묘사", "감정 표현"], "typical_scenes": ["집", "학교", "회사", "동네", "편의점"] }, "개그": { "key_elements": ["개그", "패러디", "과장", "반전", "슬랩스틱"], "visual_styles": ["과장된 표정", "데포르메", "효과선", "말풍선 연출", "파격 구도"], "panel_types": ["과장 리액션", "개그 컷", "패러디", "충격 표정", "망가짐"], "typical_scenes": ["개그 상황", "일상 붕괴", "오해 상황", "추격전", "단체 개그"] }, "스포츠": { "key_elements": ["경기", "훈련", "팀워크", "라이벌", "성장"], "visual_styles": ["다이나믹", "스피드선", "땀 표현", "근육 묘사", "경기장"], "panel_types": ["액션 컷", "결정적 순간", "전신 동작", "표정 클로즈업", "경기 전경"], "typical_scenes": ["경기장", "훈련장", "라커룸", "벤치", "시상대"] } } # --- Data classes --- @dataclass class CharacterProfile: """Character profile with celebrity lookalike""" name: str role: str personality: str appearance: str celebrity_lookalike: str gender: str @dataclass class WebtoonBible: """Webtoon story bible for maintaining consistency""" genre: str = "" title: str = "" characters: Dict[str, CharacterProfile] = field(default_factory=dict) settings: Dict[str, str] = field(default_factory=dict) plot_points: List[Dict[str, Any]] = field(default_factory=list) episode_hooks: Dict[int, str] = field(default_factory=dict) genre_elements: Dict[str, Any] = field(default_factory=dict) visual_style: Dict[str, Any] = field(default_factory=dict) panel_compositions: List[str] = field(default_factory=list) @dataclass class StoryboardPanel: """Individual storyboard panel with unique ID""" panel_number: int scene_type: str # wide, close-up, medium, establishing image_prompt: str # Image generation prompt with character descriptions panel_id: str = "" # Unique panel identifier dialogue: List[str] = field(default_factory=list) narration: str = "" sound_effects: List[str] = field(default_factory=list) emotion_notes: str = "" camera_angle: str = "" background: str = "" characters_in_scene: List[str] = field(default_factory=list) generated_image_url: str = "" # URL of generated image @dataclass class EpisodeStoryboard: """Complete storyboard for one episode""" episode_number: int title: str panels: List[StoryboardPanel] = field(default_factory=list) total_panels: int = 30 cliffhanger: str = "" # --- Core logic classes --- class WebtoonTracker: """Webtoon narrative and storyboard tracker""" def __init__(self): self.story_bible = WebtoonBible() self.episode_storyboards: Dict[int, EpisodeStoryboard] = {} self.episodes: Dict[int, str] = {} self.total_panel_count = 0 self.character_profiles: Dict[str, CharacterProfile] = {} def set_genre(self, genre: str): """Set the webtoon genre""" self.story_bible.genre = genre self.story_bible.genre_elements = GENRE_ELEMENTS.get(genre, {}) def add_character(self, character: CharacterProfile): """Add character with celebrity lookalike""" self.character_profiles[character.name] = character self.story_bible.characters[character.name] = character def add_storyboard(self, episode_num: int, storyboard: EpisodeStoryboard): """Add episode storyboard""" self.episode_storyboards[episode_num] = storyboard self.total_panel_count += len(storyboard.panels) class WebtoonDatabase: """Database management for webtoon system""" @staticmethod def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute("PRAGMA journal_mode=WAL") cursor = conn.cursor() # Sessions table with genre cursor.execute(''' CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_query TEXT NOT NULL, genre TEXT NOT NULL, language TEXT NOT NULL, title TEXT, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')), status TEXT DEFAULT 'active', current_episode INTEGER DEFAULT 0, total_episodes INTEGER DEFAULT 40, planning_doc TEXT, story_bible TEXT, visual_style TEXT, character_profiles TEXT ) ''') # Storyboards table cursor.execute(''' CREATE TABLE IF NOT EXISTS storyboards ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, episode_number INTEGER NOT NULL, title TEXT, storyboard_data TEXT, panel_count INTEGER DEFAULT 30, status TEXT DEFAULT 'pending', created_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (session_id) REFERENCES sessions(session_id), UNIQUE(session_id, episode_number) ) ''') # Panels table with image data cursor.execute(''' CREATE TABLE IF NOT EXISTS panels ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, episode_number INTEGER NOT NULL, panel_number INTEGER NOT NULL, scene_type TEXT, image_prompt TEXT, dialogue TEXT, narration TEXT, sound_effects TEXT, generated_image 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, genre: str, language: str) -> str: session_id = hashlib.md5(f"{user_query}{genre}{datetime.now()}".encode()).hexdigest() with WebtoonDatabase.get_db() as conn: conn.cursor().execute( '''INSERT INTO sessions (session_id, user_query, genre, language) VALUES (?, ?, ?, ?)''', (session_id, user_query, genre, language) ) conn.commit() return session_id @staticmethod def save_storyboard(session_id: str, episode_num: int, storyboard: EpisodeStoryboard): with WebtoonDatabase.get_db() as conn: cursor = conn.cursor() # Save storyboard cursor.execute(''' INSERT INTO storyboards (session_id, episode_number, title, storyboard_data, panel_count, status) VALUES (?, ?, ?, ?, ?, 'complete') ON CONFLICT(session_id, episode_number) DO UPDATE SET title=?, storyboard_data=?, panel_count=?, status='complete' ''', (session_id, episode_num, storyboard.title, json.dumps(asdict(storyboard)), len(storyboard.panels), storyboard.title, json.dumps(asdict(storyboard)), len(storyboard.panels))) # Save individual panels for panel in storyboard.panels: cursor.execute(''' INSERT INTO panels (session_id, episode_number, panel_number, scene_type, image_prompt, dialogue, narration, sound_effects) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ''', (session_id, episode_num, panel.panel_number, panel.scene_type, panel.image_prompt, json.dumps(panel.dialogue), panel.narration, json.dumps(panel.sound_effects))) conn.commit() @staticmethod def save_character_profiles(session_id: str, profiles: Dict[str, CharacterProfile]): with WebtoonDatabase.get_db() as conn: cursor = conn.cursor() profiles_json = json.dumps({name: asdict(profile) for name, profile in profiles.items()}) cursor.execute( "UPDATE sessions SET character_profiles = ? WHERE session_id = ?", (profiles_json, session_id) ) conn.commit() # --- Image Generation --- class ImageGenerator: """Handle image generation using Replicate API with multi-threading""" def __init__(self): self.generation_lock = Lock() self.active_generations = {} def generate_image(self, prompt: str, panel_id: str, session_id: str, progress_callback=None) -> Dict[str, Any]: """Generate image using Replicate API with progress callback""" try: if not REPLICATE_API_TOKEN: logger.warning("No Replicate API token, returning placeholder") return {"panel_id": panel_id, "status": "error", "message": "No API token"} logger.info(f"Generating image for panel {panel_id}") if progress_callback: progress_callback(f"패널 {panel_id.split('_panel')[1]} 이미지 생성 중...") # Run the model input_params = { "prompt": prompt, "num_inference_steps": 25, "guidance_scale": 7.5 } output = replicate.run( "stability-ai/sdxl:39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b", input=input_params ) # Get the image URL if output and len(output) > 0: image_url = output[0] if isinstance(output[0], str) else str(output[0]) # Cache the image cache_key = f"{session_id}_{panel_id}" generated_images_cache[cache_key] = image_url logger.info(f"Successfully generated image for panel {panel_id}") return { "panel_id": panel_id, "status": "success", "image_url": image_url, "prompt": prompt } return {"panel_id": panel_id, "status": "error", "message": "No output from model"} except Exception as e: logger.error(f"Image generation error for panel {panel_id}: {e}") return {"panel_id": panel_id, "status": "error", "message": str(e)} def generate_multiple_images(self, panel_prompts: List[Dict], session_id: str, max_workers: int = 3, progress_callback=None) -> List[Dict]: """Generate multiple images in parallel with progress tracking""" results = [] total = len(panel_prompts) completed = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: # Submit all tasks future_to_panel = {} for i, panel_data in enumerate(panel_prompts): panel_id = panel_data['panel_id'] prompt = panel_data['prompt'] future = executor.submit( self.generate_image, prompt, panel_id, session_id, None # Individual progress not tracked in parallel ) future_to_panel[future] = panel_data # Collect results as they complete for future in concurrent.futures.as_completed(future_to_panel): panel_data = future_to_panel[future] try: result = future.result(timeout=60) results.append(result) completed += 1 if progress_callback: progress_callback(f"진행중: {completed}/{total} 패널 완료") except concurrent.futures.TimeoutError: results.append({ "panel_id": panel_data['panel_id'], "status": "error", "message": "Generation timeout" }) completed += 1 except Exception as e: results.append({ "panel_id": panel_data['panel_id'], "status": "error", "message": str(e) }) completed += 1 # Sort results by panel_id to maintain order results.sort(key=lambda x: int(x['panel_id'].split('_panel')[1])) return results # --- LLM Integration --- class WebtoonSystem: """Webtoon planning and storyboard generation system""" def __init__(self): self.api_key = FIREWORKS_API_KEY self.api_url = API_URL self.model_id = MODEL_ID self.tracker = WebtoonTracker() self.current_session_id = None self.image_generator = ImageGenerator() WebtoonDatabase.init_db() def create_headers(self): return { "Accept": "application/json", "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } def assign_celebrity_lookalikes(self, characters: List[Dict]) -> Dict[str, CharacterProfile]: """Assign celebrity lookalikes to characters""" profiles = {} used_celebrities = [] for char in characters: gender = char.get('gender', 'male') available_celebrities = [c for c in CELEBRITY_FACES.get(gender, []) if c not in used_celebrities] if not available_celebrities: available_celebrities = CELEBRITY_FACES.get(gender, []) celebrity = random.choice(available_celebrities) used_celebrities.append(celebrity) profile = CharacterProfile( name=char.get('name', ''), role=char.get('role', ''), personality=char.get('personality', ''), appearance=char.get('appearance', ''), celebrity_lookalike=celebrity, gender=gender ) profiles[profile.name] = profile self.tracker.add_character(profile) return profiles # --- Prompt generation functions --- def create_planning_prompt(self, query: str, genre: str, language: str) -> str: """Create initial planning prompt for webtoon with character profiles""" genre_info = GENRE_ELEMENTS.get(genre, {}) lang_prompts = { "Korean": f"""한국 웹툰 시장을 겨냥한 {genre} 장르 웹툰을 기획하세요. **[핵심 스토리 설정 - 반드시 이 내용을 중심으로 전개하세요]** {query} **장르:** {genre} **목표:** 40화 완결 웹툰 ⚠️ **중요**: 1. 위에 제시된 스토리 설정을 반드시 기반으로 하여 플롯을 구성하세요. 2. 각 캐릭터의 성별(gender)을 명확히 지정하세요 (male/female). 3. 반드시 40화 전체 구성안을 모두 작성하세요. **장르 필수 요소:** - 핵심 요소: {', '.join(genre_info.get('key_elements', []))} - 비주얼 스타일: {', '.join(genre_info.get('visual_styles', []))} - 주요 씬: {', '.join(genre_info.get('typical_scenes', []))} 다음 형식으로 작성하세요: 📚 **작품 제목:** [임팩트 있는 제목] 🎨 **비주얼 컨셉:** - 그림체: [작품에 어울리는 그림체] - 색감: [주요 색상 톤] - 캐릭터 디자인 특징: [주인공들의 비주얼 특징] 👥 **주요 캐릭터:** (각 캐릭터마다 성별을 반드시 명시!) - 주인공: [이름] - 성별: [male/female] - [외모 특징, 성격, 목표] - 캐릭터2: [이름] - 성별: [male/female] - [역할, 특징] - 캐릭터3: [이름] - 성별: [male/female] - [역할, 특징] 📖 **시놉시스:** [3-4줄로 전체 스토리 요약] 📝 **40화 전체 구성안:** (반드시 40화 모두 작성!) 각 화별로 핵심 사건과 클리프행어를 포함하여 작성하세요. 1화: [제목] - [핵심 사건] - 클리프행어: [충격적인 마무리] 2화: [제목] - [핵심 사건] - 클리프행어: [충격적인 마무리] 3화: [제목] - [핵심 사건] - 클리프행어: [충격적인 마무리] ... (중간 생략하지 말고 모든 화를 작성) ... 38화: [제목] - [핵심 사건] - 클리프행어: [충격적인 마무리] 39화: [제목] - [핵심 사건] - 클리프행어: [충격적인 마무리] 40화: [제목] - [핵심 사건] - [대단원의 마무리] ⚠️ 절대 생략하지 말고 40화 모두 작성하세요!""", "English": f"""Plan a Korean-style webtoon for {genre} genre. **[Core Story Setting - MUST base the story on this]** {query} **Genre:** {genre} **Goal:** 40 episodes webtoon ⚠️ **IMPORTANT**: 1. You MUST base the plot on the story setting provided above. 2. Clearly specify each character's gender (male/female). 3. MUST write all 40 episodes structure. **Genre Requirements:** - Key elements: {', '.join(genre_info.get('key_elements', []))} - Visual styles: {', '.join(genre_info.get('visual_styles', []))} - Typical scenes: {', '.join(genre_info.get('typical_scenes', []))} Format as follows: 📚 **Title:** [Impactful title] 🎨 **Visual Concept:** - Art style: [Suitable art style] - Color tone: [Main color palette] - Character design: [Visual characteristics] 👥 **Main Characters:** (Must specify gender for each!) - Protagonist: [Name] - Gender: [male/female] - [Appearance, personality, goal] - Character2: [Name] - Gender: [male/female] - [Role, traits] - Character3: [Name] - Gender: [male/female] - [Role, traits] 📖 **Synopsis:** [3-4 line story summary] 📝 **40 Episode Structure:** (MUST write all 40 episodes!) Include key events and cliffhangers for each episode. Episode 1: [Title] - [Key event] - Cliffhanger: [Shocking ending] Episode 2: [Title] - [Key event] - Cliffhanger: [Shocking ending] ... (Don't skip, write all episodes) ... Episode 40: [Title] - [Key event] - [Grand finale] ⚠️ Don't abbreviate, write all 40 episodes!""" } return lang_prompts.get(language, lang_prompts["Korean"]) def create_storyboard_prompt(self, episode_num: int, plot_outline: str, genre: str, language: str, character_profiles: Dict[str, CharacterProfile]) -> str: """Create prompt for episode storyboard with character descriptions""" genre_info = GENRE_ELEMENTS.get(genre, {}) # Create character description string char_descriptions = "\n".join([ f"- {name}: {profile.celebrity_lookalike} 닮은 얼굴의 {profile.gender}" for name, profile in character_profiles.items() ]) lang_prompts = { "Korean": f"""웹툰 {episode_num}화 스토리보드를 30개 패널로 작성하세요. **장르:** {genre} **1화 내용:** {self._extract_episode_plan(plot_outline, episode_num)} **캐릭터 얼굴 설정:** {char_descriptions} ⚠️ **중요**: 1. 반드시 30개 패널을 모두 작성하세요! 2. 캐릭터가 등장할 때마다 "캐릭터이름(유명인 닮은 얼굴의 성별)" 형식으로 작성하세요! 예시: "홍길동(톰 크루즈 닮은 얼굴의 남자)이 거리를 걷고 있다" **패널 구성 지침:** - 총 30개 패널로 구성 - 다양한 샷 사이즈 활용 - 장르 특성에 맞는 연출: {', '.join(genre_info.get('panel_types', []))} **각 패널별로 다음을 포함하여 작성:** 패널 1: - 샷 타입: [establishing/wide/medium/close_up 등] - 이미지 프롬프트: [캐릭터 설명 포함한 상세한 한글 이미지 생성 프롬프트] - 대사: [캐릭터 대사가 있다면] - 나레이션: [해설이 있다면] - 효과음: [필요한 효과음] - 배경: [배경 설명] 패널 2: (위와 같은 형식) ...이런 식으로 30개 패널 모두 작성 패널 30: (위와 같은 형식) ⚠️ 반드시 30개 패널을 모두 작성하세요. 생략하지 마세요!""", "English": f"""Create Episode {episode_num} storyboard with 30 panels. **Genre:** {genre} **Episode content:** {self._extract_episode_plan(plot_outline, episode_num)} **Character Face Settings:** {char_descriptions} ⚠️ **IMPORTANT**: 1. Must write all 30 panels! 2. Always describe characters as "CharacterName (celebrity lookalike face gender)"! Example: "John (Tom Cruise lookalike male) walking down the street" **Panel composition guidelines:** - Total 30 panels - Various shot sizes - Genre-appropriate directing: {', '.join(genre_info.get('panel_types', []))} **For each panel include:** Panel 1: - Shot type: [establishing/wide/medium/close_up etc] - Image prompt: [Detailed prompt with character descriptions] - Dialogue: [Character dialogue if any] - Narration: [Narration if any] - Sound effects: [Required sound effects] - Background: [Background description] Panel 2: (Same format) ...continue for all 30 panels Panel 30: (Same format) ⚠️ Must write all 30 panels. Don't skip any!""" } return lang_prompts.get(language, lang_prompts["Korean"]) def _extract_episode_plan(self, plot_outline: str, episode_num: int) -> str: """Extract specific episode plan from outline""" lines = plot_outline.split('\n') episode_section = [] capturing = False patterns = [ f"{episode_num}화:", f"Episode {episode_num}:", f"제{episode_num}화:", f"EP{episode_num}:", f"{episode_num}.", f"[{episode_num}]" ] next_patterns = [ f"{episode_num+1}화:", f"Episode {episode_num+1}:", f"제{episode_num+1}화:", f"EP{episode_num+1}:", f"{episode_num+1}.", f"[{episode_num+1}]" ] for line in lines: if any(pattern in line for pattern in patterns): capturing = True episode_section.append(line) elif capturing and any(pattern in line for pattern in next_patterns): break elif capturing: episode_section.append(line) if episode_section: return '\n'.join(episode_section) return f"에피소드 {episode_num} 내용을 플롯에서 참조하여 작성하세요." def parse_characters_from_planning(self, planning_doc: str) -> List[Dict]: """Parse character information from planning document""" characters = [] lines = planning_doc.split('\n') in_character_section = False current_char = {} for line in lines: if '주요 캐릭터' in line or 'Main Characters' in line: in_character_section = True continue elif in_character_section and ('시놉시스' in line or 'Synopsis' in line): if current_char: characters.append(current_char) break elif in_character_section and line.strip(): # Parse character line if '성별:' in line or 'Gender:' in line: if current_char: characters.append(current_char) parts = line.split('-') if len(parts) >= 2: name = parts[0].strip().replace('주인공:', '').replace('캐릭터', '').strip() # Extract gender gender = 'male' # default if 'female' in line.lower() or '여' in line: gender = 'female' elif 'male' in line.lower() or '남' in line: gender = 'male' current_char = { 'name': name, 'gender': gender, 'role': parts[1].strip() if len(parts) > 1 else '', 'personality': parts[2].strip() if len(parts) > 2 else '', 'appearance': parts[3].strip() if len(parts) > 3 else '' } if current_char and current_char not in characters: characters.append(current_char) # Ensure at least 3 characters while len(characters) < 3: characters.append({ 'name': f'캐릭터{len(characters)+1}', 'gender': 'male' if len(characters) % 2 == 0 else 'female', 'role': '조연', 'personality': '일반적', 'appearance': '평범한 외모' }) return characters # --- LLM call functions --- 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 == "storyboarder" else 10000 payload = { "model": self.model_id, "messages": full_messages, "max_tokens": max_tokens, "temperature": 0.7, "top_p": 1, "top_k": 40, "presence_penalty": 0, "frequency_penalty": 0, "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 Error (Status Code: {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"Chunk processing error: {str(e)}") continue if buffer: yield buffer except Exception as e: logger.error(f"Streaming error: {type(e).__name__}: {str(e)}") yield f"❌ Error occurred: {str(e)}" def get_system_prompts(self, language: str) -> Dict[str, str]: """System prompts for webtoon roles""" base_prompts = { "Korean": { "planner": """당신은 한국 웹툰 시장을 완벽히 이해하는 웹툰 기획자입니다. 독자를 사로잡는 스토리와 비주얼 연출을 기획합니다. 40화 완결 구조로 완벽한 기승전결을 설계합니다. 각 화마다 강력한 클리프행어로 다음 화를 기대하게 만듭니다. 캐릭터의 성별을 명확히 지정합니다. ⚠️ 가장 중요한 원칙: 1. 사용자가 제공한 스토리 설정을 절대적으로 우선시하고, 이를 중심으로 모든 플롯을 구성합니다. 2. 반드시 40화 전체 구성안을 모두 작성합니다. 생략하지 않습니다.""", "storyboarder": """당신은 웹툰 스토리보드 전문가입니다. 30개 패널로 한 화를 완벽하게 구성합니다. 세로 스크롤에 최적화된 연출을 합니다. 각 패널마다 캐릭터의 유명인 닮은꼴 설정을 포함한 상세한 이미지 프롬프트를 작성합니다. ⚠️ 가장 중요한 원칙: 1. 반드시 30개 패널을 모두 작성합니다. 2. 캐릭터가 등장할 때마다 "캐릭터이름(유명인 닮은 얼굴의 성별)" 형식으로 작성합니다.""" }, "English": { "planner": """You perfectly understand the Korean webtoon market. Design stories and visual direction that captivate readers. Create perfect story structure in 40 episodes. Make readers anticipate next episode with strong cliffhangers. Clearly specify character genders. ⚠️ Most important principles: 1. Absolutely prioritize the user's story setting and build all plots around it. 2. Must write all 40 episodes structure. Don't skip.""", "storyboarder": """You are a webtoon storyboard specialist. Perfectly compose one episode with 30 panels. Write detailed image prompts including celebrity lookalike descriptions. ⚠️ Most important principles: 1. Must write all 30 panels. 2. Always describe characters as "CharacterName (celebrity lookalike face gender)".""" } } return base_prompts.get(language, base_prompts["Korean"]) # --- Main process --- def process_webtoon_stream(self, query: str, genre: str, language: str, session_id: Optional[str] = None) -> Generator[Tuple[str, str, str, str, Dict], None, None]: """Webtoon planning and storyboard generation process""" try: if not session_id: self.current_session_id = WebtoonDatabase.create_session(query, genre, language) self.tracker.set_genre(genre) logger.info(f"Created new session: {self.current_session_id}") self.original_query = query else: self.current_session_id = session_id # Phase 1: Generate planning document (40화 구성 포함) yield "🎬 웹툰 기획안 작성 중... (40화 전체 구성 포함)", "", f"장르: {genre}", self.current_session_id, {} planning_prompt = self.create_planning_prompt(query, genre, language) planning_doc = self.call_llm_sync( [{"role": "user", "content": planning_prompt}], "planner", language ) self.planning_doc = planning_doc # Parse characters and assign celebrity lookalikes characters = self.parse_characters_from_planning(planning_doc) character_profiles = self.assign_celebrity_lookalikes(characters) # Save character profiles WebtoonDatabase.save_character_profiles(self.current_session_id, character_profiles) yield "✅ 기획안 완성! (40화 구성 완료)", planning_doc, "40화 구성 완료", self.current_session_id, character_profiles # Phase 2: Generate Episode 1 Storyboard yield "🎨 1화 스토리보드 작성 중... (30개 패널)", planning_doc, "30개 패널 구성 중", self.current_session_id, character_profiles storyboard_prompt = self.create_storyboard_prompt(1, planning_doc, genre, language, character_profiles) storyboard_content = self.call_llm_sync( [{"role": "user", "content": storyboard_prompt}], "storyboarder", language ) # Parse storyboard into structured format storyboard = self.parse_storyboard(storyboard_content, 1, character_profiles) # Save to database WebtoonDatabase.save_storyboard(self.current_session_id, 1, storyboard) yield "🎉 완성! (기획안 + 1화 스토리보드)", planning_doc, storyboard_content, self.current_session_id, character_profiles except Exception as e: logger.error(f"Webtoon generation error: {e}", exc_info=True) yield f"❌ 오류 발생: {e}", "", "", self.current_session_id, {} def parse_storyboard(self, content: str, episode_num: int, character_profiles: Dict[str, CharacterProfile]) -> EpisodeStoryboard: """Parse storyboard text into structured format with unique panel IDs""" storyboard = EpisodeStoryboard(episode_number=episode_num, title=f"{episode_num}화") panels = [] current_panel = None panel_number = 0 lines = content.split('\n') for line in lines: if '패널' in line or 'Panel' in line: if current_panel: panels.append(current_panel) panel_number += 1 # Create unique panel ID panel_id = f"ep{episode_num}_panel{panel_number}" current_panel = StoryboardPanel( panel_number=panel_number, scene_type="medium", image_prompt="", panel_id=panel_id # Add unique panel_id ) elif current_panel: if '이미지 프롬프트:' in line or 'Image prompt:' in line: current_panel.image_prompt = line.split(':', 1)[1].strip() elif '대사:' in line or 'Dialogue:' in line: dialogue = line.split(':', 1)[1].strip() if dialogue: current_panel.dialogue.append(dialogue) elif '나레이션:' in line or 'Narration:' in line: current_panel.narration = line.split(':', 1)[1].strip() elif '효과음:' in line or 'Sound effects:' in line: effects = line.split(':', 1)[1].strip() if effects: current_panel.sound_effects.append(effects) elif '샷 타입:' in line or 'Shot type:' in line: current_panel.scene_type = line.split(':', 1)[1].strip() if current_panel: panels.append(current_panel) storyboard.panels = panels[:30] return storyboard # --- Format storyboard for display --- def format_storyboard_for_display(storyboard_content: str, character_profiles: Dict[str, CharacterProfile], session_id: str) -> str: """Format storyboard content for panel display""" formatted_panels = [] panel_texts = [] current_panel = {} lines = storyboard_content.split('\n') for line in lines: if '패널' in line or 'Panel' in line: if current_panel: panel_texts.append(current_panel) current_panel = {'number': len(panel_texts) + 1} elif '이미지 프롬프트:' in line or 'Image prompt:' in line: current_panel['prompt'] = line.split(':', 1)[1].strip() elif '대사:' in line or 'Dialogue:' in line: current_panel['dialogue'] = line.split(':', 1)[1].strip() elif '나레이션:' in line or 'Narration:' in line: current_panel['narration'] = line.split(':', 1)[1].strip() elif '효과음:' in line or 'Sound effects:' in line: current_panel['effects'] = line.split(':', 1)[1].strip() elif '샷 타입:' in line or 'Shot type:' in line: current_panel['shot'] = line.split(':', 1)[1].strip() if current_panel: panel_texts.append(current_panel) # Format each panel for display for panel in panel_texts[:30]: panel_num = panel.get('number', '?') panel_html = f"""

🎬 패널 {panel_num}

샷 타입: {panel.get('shot', 'N/A')}

이미지 프롬프트:
{panel.get('prompt', 'N/A')}

{f"

대사: {panel.get('dialogue')}

" if panel.get('dialogue') else ""} {f"

나레이션: {panel.get('narration')}

" if panel.get('narration') else ""} {f"

효과음: {panel.get('effects')}

" if panel.get('effects') else ""}
""" formatted_panels.append(panel_html) return ''.join(formatted_panels) # --- Export functions --- def export_to_txt(planning_doc: str, storyboard: str, genre: str, title: str = "") -> str: """Export webtoon planning and storyboard to TXT format""" content = f"{'=' * 50}\n" content += f"{title if title else genre + ' 웹툰'}\n" content += f"{'=' * 50}\n\n" content += f"장르: {genre}\n" content += f"총 40화 기획\n" content += f"{'=' * 50}\n\n" content += "📚 기획안 (40화 전체 구성)\n\n" content += planning_doc content += f"\n\n{'=' * 50}\n\n" content += "🎨 1화 스토리보드 (30 패널)\n\n" content += storyboard return content def generate_random_webtoon_theme(genre: str, language: str) -> str: """Generate random webtoon theme""" templates = { "로맨스": [ "재벌 3세 상사와 신입사원의 비밀 계약연애", "고등학교 때 첫사랑과 10년 만의 재회", "냉혈 검사와 열혈 변호사의 법정 로맨스" ], "로판": [ "악녀로 빙의했는데 1년 후 처형 예정", "회귀한 황녀, 버려진 왕자와 손잡다", "계약결혼한 북부 공작이 집착남이 되었다" ], "판타지": [ "F급 헌터가 SSS급 네크로맨서로 각성", "100층 탑을 역주행하는 회귀자", "버그로 최강이 된 게임 속 NPC" ], "현판": [ "무능력자인 줄 알았는데 SSS급 생산직", "게이트 속에서 10년, 돌아온 최강자", "헌터 고등학교의 숨겨진 랭킹 1위" ], "무협": [ "천하제일문 막내가 마교 교주 제자가 되다", "100년 전으로 회귀한 화산파 장문인", "폐급 무공으로 천하를 제패하다" ], "스릴러": [ "폐교에 갇힌 동창회, 한 명씩 사라진다", "타임루프 속 연쇄살인범 찾기", "내 남편이 사이코패스였다" ], "일상": [ "편의점 알바생의 소소한 일상", "30대 직장인의 퇴사 준비 일기", "우리 동네 고양이들의 비밀 회의" ], "개그": [ "이세계 용사인데 스탯이 이상하다", "우리 학교 선생님은 전직 마왕", "좀비 아포칼립스인데 나만 개그 캐릭터" ], "스포츠": [ "벤치 멤버에서 에이스가 되기까지", "여자 야구부 창설기", "은퇴 선수의 코치 도전기" ] } genre_themes = templates.get(genre, templates["로맨스"]) return random.choice(genre_themes) # --- Gradio interface --- def create_interface(): with gr.Blocks(theme=gr.themes.Soft(), title="K-Webtoon Storyboard Generator") as interface: gr.HTML("""

🎨 K-Webtoon Storyboard Generator

한국형 웹툰 기획 및 스토리보드 자동 생성 시스템

📚 40화 전체 구성 + 🎬 1화 30패널 스토리보드

""") # State current_session_id = gr.State(None) planning_state = gr.State("") storyboard_state = gr.State("") character_profiles_state = gr.State({}) with gr.Tab("📚 기획안 작성 (40화 구성)"): with gr.Group(): gr.Markdown("### 🎯 웹툰 설정") with gr.Row(): with gr.Column(scale=2): genre_select = gr.Radio( choices=list(WEBTOON_GENRES.keys()), value="로맨스", label="장르 선택", info="원하는 장르를 선택하세요" ) query_input = gr.Textbox( label="스토리 콘셉트", placeholder="웹툰의 기본 설정이나 주제를 입력하세요...", lines=3 ) with gr.Row(): random_btn = gr.Button("🎲 랜덤 테마", variant="secondary") submit_btn = gr.Button("📝 기획 시작", variant="primary", size="lg") with gr.Column(scale=1): language_select = gr.Radio( choices=["Korean", "English"], value="Korean", label="언어" ) gr.Markdown(""" **🎭 캐릭터 얼굴 설정** 각 캐릭터에 유명인 닮은꼴이 자동으로 배정됩니다. **📋 생성 내용** - 40화 전체 구성안 - 각 화별 클리프행어 - 캐릭터 프로필 - 1화 30패널 스토리보드 """) status_text = gr.Textbox( label="진행 상황", interactive=False, value="장르를 선택하고 콘셉트를 입력하세요" ) # Planning output with better visibility for 40 episodes gr.Markdown("### 📖 웹툰 기획안 (40화 전체 구성)") planning_display = gr.Markdown("*기획안이 여기에 표시됩니다 (40화 전체 구성 포함)*") # Character profiles display character_display = gr.HTML(label="캐릭터 프로필") with gr.Row(): download_format = gr.Radio( choices=["TXT"], value="TXT", label="다운로드 형식" ) download_btn = gr.Button("📥 다운로드", variant="secondary") download_file = gr.File(visible=False) with gr.Tab("🎬 1화 스토리보드 (30패널)"): gr.Markdown(""" ### 📋 1화 스토리보드 (30 패널) 좌측: 텍스트 및 이미지 프롬프트 | 우측: 생성된 이미지 """) # Progress bar for image generation progress_bar = gr.Progress(visible=False) generation_status = gr.Textbox( label="이미지 생성 상태", value="대기 중...", interactive=False, visible=False ) with gr.Row(): with gr.Column(): storyboard_text_display = gr.HTML( value="

기획안 작성 후 스토리보드가 생성됩니다

", label="패널 텍스트 & 프롬프트" ) with gr.Column(): image_display_area = gr.HTML( value="""

🖼️ 이미지 생성 공간

각 패널의 이미지를 생성하려면 아래 버튼을 클릭하세요

""", label="생성된 이미지" ) # Image generation controls with progress with gr.Row(): generate_all_btn = gr.Button("🎨 모든 이미지 생성 (30개 패널)", variant="primary") clear_images_btn = gr.Button("🗑️ 이미지 초기화", variant="secondary") generation_progress = gr.Number(value=0, label="생성 진행률 (%)", visible=False) # Event handlers def process_query(query, genre, language, session_id): system = WebtoonSystem() planning = "" storyboard = "" character_profiles = {} for planning_content, storyboard_content, status, new_session_id, profiles in system.process_webtoon_stream(query, genre, language, session_id): planning = planning_content storyboard = storyboard_content character_profiles = profiles yield planning, storyboard, status, new_session_id, character_profiles def format_character_profiles(profiles: Dict[str, CharacterProfile]) -> str: """Format character profiles for display""" if not profiles: return "" html = "

🎭 캐릭터 프로필

" for name, profile in profiles.items(): html += f"""
{name} - {profile.celebrity_lookalike} 닮은 얼굴
역할: {profile.role}
성격: {profile.personality}
""" return html def update_storyboard_display(storyboard_content, character_profiles, session_id): """Update storyboard display with formatted panels""" if not storyboard_content or storyboard_content == "": return "

기획안 작성 후 스토리보드가 생성됩니다

" return format_storyboard_for_display(storyboard_content, character_profiles, session_id) def handle_random_theme(genre, language): return generate_random_webtoon_theme(genre, language) def handle_download(download_format, session_id, planning, storyboard, genre): """Handle download request""" try: title = f"{genre} 웹툰" content = export_to_txt(planning, storyboard, genre, title) with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', suffix='.txt', delete=False) as f: f.write(content) return f.name except Exception as e: logger.error(f"Download error: {e}") gr.Warning(f"다운로드 중 오류 발생: {str(e)}") return None def generate_all_images_with_progress(session_id, storyboard_content, character_profiles, progress=gr.Progress()): """Generate images for all panels with progress tracking""" if not REPLICATE_API_TOKEN: return "

⚠️ Replicate API 토큰이 설정되지 않았습니다.

" if not storyboard_content: return "

⚠️ 먼저 스토리보드를 생성하세요.

" # Parse storyboard to extract prompts panel_prompts = [] lines = storyboard_content.split('\n') current_panel_num = 0 current_prompt = "" for line in lines: if '패널' in line or 'Panel' in line: if current_prompt and current_panel_num > 0: panel_prompts.append({ 'panel_id': f"ep1_panel{current_panel_num}", 'panel_num': current_panel_num, 'prompt': current_prompt }) current_panel_num += 1 current_prompt = "" elif '이미지 프롬프트:' in line or 'Image prompt:' in line: current_prompt = line.split(':', 1)[1].strip() # Add last panel if exists if current_prompt and current_panel_num > 0: panel_prompts.append({ 'panel_id': f"ep1_panel{current_panel_num}", 'panel_num': current_panel_num, 'prompt': current_prompt }) # Limit to 30 panels panel_prompts = panel_prompts[:30] total_panels = len(panel_prompts) if total_panels == 0: return "

⚠️ 패널 정보를 찾을 수 없습니다.

" # Initialize progress progress(0, desc=f"이미지 생성 준비 중... (총 {total_panels}개 패널)") # Generate images with progress tracking image_generator = ImageGenerator() image_html = "
" image_html += f"

🖼️ 이미지 생성 결과

" results = [] # Generate images one by one with progress updates for i, panel_data in enumerate(panel_prompts): progress((i / total_panels), desc=f"패널 {panel_data['panel_num']}/{total_panels} 생성 중...") try: result = image_generator.generate_image( panel_data['prompt'], panel_data['panel_id'], session_id ) results.append(result) except Exception as e: results.append({ "panel_id": panel_data['panel_id'], "status": "error", "message": str(e) }) # Small delay to prevent API rate limiting time.sleep(0.5) progress(1.0, desc="이미지 생성 완료!") # Display results success_count = sum(1 for r in results if r['status'] == 'success') image_html += f"

✅ 성공: {success_count}/{len(results)} 패널

" # Grid layout for images image_html += "
" for result in results: panel_num = int(result['panel_id'].split('_panel')[1]) if result['status'] == 'success': image_html += f"""

패널 {panel_num}

프롬프트 보기

{result.get('prompt', '')[:150]}...

""" else: image_html += f"""

패널 {panel_num} - 생성 실패

❌ {result.get('message', 'Unknown error')}

""" image_html += "
" return image_html def clear_all_images(): """Clear all generated images""" return """

🖼️ 이미지 생성 공간

이미지가 초기화되었습니다

""" # Connect events submit_btn.click( fn=process_query, inputs=[query_input, genre_select, language_select, current_session_id], outputs=[planning_state, storyboard_state, status_text, current_session_id, character_profiles_state] ).then( fn=lambda x: x, inputs=[planning_state], outputs=[planning_display] ).then( fn=format_character_profiles, inputs=[character_profiles_state], outputs=[character_display] ).then( fn=update_storyboard_display, inputs=[storyboard_state, character_profiles_state, current_session_id], outputs=[storyboard_text_display] ) random_btn.click( fn=handle_random_theme, inputs=[genre_select, language_select], outputs=[query_input] ) download_btn.click( fn=handle_download, inputs=[download_format, current_session_id, planning_state, storyboard_state, genre_select], outputs=[download_file] ).then( fn=lambda x: gr.update(visible=True) if x else gr.update(visible=False), inputs=[download_file], outputs=[download_file] ) generate_all_btn.click( fn=generate_all_images_with_progress, inputs=[current_session_id, storyboard_state, character_profiles_state], outputs=[image_display_area] ) clear_images_btn.click( fn=clear_all_images, inputs=[], outputs=[image_display_area] ) return interface # Main if __name__ == "__main__": logger.info("K-Webtoon Storyboard Generator Starting...") logger.info("=" * 60) # Environment check logger.info(f"API Endpoint: {API_URL}") logger.info(f"Model: {MODEL_ID}") logger.info(f"Target: {TARGET_EPISODES} episodes, {PANELS_PER_EPISODE} panels per episode") logger.info("Genres: " + ", ".join(WEBTOON_GENRES.keys())) if REPLICATE_API_TOKEN: logger.info("Replicate API: Configured ✓") else: logger.warning("Replicate API: Not configured (Image generation disabled)") logger.info("=" * 60) # Initialize database logger.info("Initializing database...") WebtoonDatabase.init_db() logger.info("Database ready.") # Launch interface interface = create_interface() interface.launch( server_name="0.0.0.0", server_port=7860, share=False )