diff --git a/.env.example b/.env.example index b0396e3..91be9cf 100644 --- a/.env.example +++ b/.env.example @@ -75,3 +75,17 @@ TIKTOK_OPEN_ID= # YouTube Data API v3 (기존 Google Cloud 프로젝트에서 API 추가 활성화) # YouTube Studio > 채널 > 고급 설정에서 채널 ID 확인 YOUTUBE_CHANNEL_ID= + +# ─── v3 엔진 추상화 (선택) ──────────────────────────── +# Seedance 2.0 — AI 시네마틱 영상 생성 (소설 쇼츠 권장) +# https://seedance2.ai/ +SEEDANCE_API_KEY= +# ElevenLabs — 고품질 한국어 TTS +# https://elevenlabs.io/ +ELEVENLABS_API_KEY= +# Google Gemini — 글쓰기 대체 / Veo 영상 +# https://aistudio.google.com/ +GEMINI_API_KEY= +# Runway Gen-3 — AI 영상 생성 +# https://runwayml.com/ +RUNWAY_API_KEY= diff --git a/bots/collector_bot.py b/bots/collector_bot.py index b5252ef..009c83f 100644 --- a/bots/collector_bot.py +++ b/bots/collector_bot.py @@ -97,10 +97,24 @@ def calc_freshness_score(published_at: datetime | None, max_score: int = 20) -> def calc_korean_relevance(text: str, rules: dict) -> int: """한국 독자 관련성 점수""" + max_score = rules['scoring']['korean_relevance']['max'] keywords = rules['scoring']['korean_relevance']['keywords'] + + # 한국어 문자(가-힣) 비율 체크 — 한국어 콘텐츠 자체에 기본점수 부여 + korean_chars = sum(1 for c in text if '\uac00' <= c <= '\ud7a3') + korean_ratio = korean_chars / max(len(text), 1) + if korean_ratio >= 0.15: + base = 15 # 한국어 텍스트면 기본 15점 + elif korean_ratio >= 0.05: + base = 8 + else: + base = 0 + + # 브랜드/지역 키워드 보너스 matched = sum(1 for kw in keywords if kw in text) - score = min(matched * 6, rules['scoring']['korean_relevance']['max']) - return score + bonus = min(matched * 5, max_score - base) + + return min(base + bonus, max_score) def calc_source_trust(source_url: str, rules: dict) -> tuple[int, str]: @@ -215,9 +229,14 @@ def calculate_quality_score(item: dict, rules: dict) -> int: kr_score = calc_korean_relevance(text, rules) fresh_score = calc_freshness_score(pub_at) - # search_demand: pytrends 연동 후 실제값 사용 (현재 기본값 10) - search_score = item.get('search_demand_score', 10) - trust_score, trust_level = calc_source_trust(source_url, rules) + # search_demand: pytrends 연동 후 실제값 사용 (RSS 기본값 12) + search_score = item.get('search_demand_score', 12) + # 신뢰도: _trust_override 이미 설정된 경우 우선 사용 + if '_trust_score' in item: + trust_score = item['_trust_score'] + trust_level = item.get('source_trust_level', 'medium') + else: + trust_score, trust_level = calc_source_trust(source_url, rules) mono_score = calc_monetization(text, rules) item['korean_relevance_score'] = kr_score diff --git a/bots/converters/video_engine.py b/bots/converters/video_engine.py new file mode 100644 index 0000000..87623b4 --- /dev/null +++ b/bots/converters/video_engine.py @@ -0,0 +1,783 @@ +""" +비디오 엔진 추상화 (bots/converters/video_engine.py) +역할: engine.json video_generation 설정에 따라 적절한 영상 생성 엔진 인스턴스 반환 +설계서: blog-engine-final-masterplan-v3.txt + +지원 엔진: + - FFmpegSlidesEngine: 기존 shorts_converter.py 파이프라인 (슬라이드 + TTS + ffmpeg) + - SeedanceEngine: Seedance 2.0 API (AI 영상 생성) + - SoraEngine: OpenAI Sora (미지원 → ffmpeg_slides 폴백) + - RunwayEngine: Runway Gen-3 API + - VeoEngine: Google Veo 3.1 (미지원 → ffmpeg_slides 폴백) +""" +import json +import logging +import os +import shutil +import subprocess +import tempfile +from abc import ABC, abstractmethod +from datetime import datetime +from pathlib import Path +from typing import Optional + +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = Path(__file__).parent.parent.parent +LOG_DIR = BASE_DIR / 'logs' +OUTPUT_DIR = BASE_DIR / 'data' / 'outputs' +ASSETS_DIR = BASE_DIR / 'assets' +BGM_PATH = ASSETS_DIR / 'bgm.mp3' + +LOG_DIR.mkdir(exist_ok=True) +OUTPUT_DIR.mkdir(exist_ok=True) + +logger = logging.getLogger(__name__) +if not logger.handlers: + handler = logging.FileHandler(LOG_DIR / 'video_engine.log', encoding='utf-8') + handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')) + logger.addHandler(handler) + logger.addHandler(logging.StreamHandler()) + logger.setLevel(logging.INFO) + + +# ─── 추상 기본 클래스 ────────────────────────────────── + +class VideoEngine(ABC): + @abstractmethod + def generate(self, scenes: list, output_path: str, **kwargs) -> str: + """ + scenes로 영상 생성. + + scenes 형식: + [ + { + "text": str, # 자막/TTS 텍스트 + "type": str, # "intro"|"headline"|"point"|"data"|"outro" + "image_prompt": str, # DALL-E 배경 프롬프트 (선택) + "slide_path": str, # 슬라이드 PNG 경로 (있으면 사용) + "audio_path": str, # TTS WAV 경로 (있으면 사용) + } + ] + + Returns: 생성된 MP4 파일 경로 (실패 시 빈 문자열) + """ + + +# ─── FFmpegSlidesEngine ──────────────────────────────── + +class FFmpegSlidesEngine(VideoEngine): + """ + 기존 shorts_converter.py의 ffmpeg 파이프라인을 재사용하는 엔진. + scenes에 slide_path + audio_path가 있으면 그대로 사용, + 없으면 빈 슬라이드와 gTTS로 생성 후 진행. + """ + + def __init__(self, cfg: dict): + self.cfg = cfg + self.ffmpeg_path = os.getenv('FFMPEG_PATH', 'ffmpeg') + self.ffprobe_path = os.getenv('FFPROBE_PATH', 'ffprobe') + self.resolution = cfg.get('resolution', '1080x1920') + self.fps = cfg.get('fps', 30) + self.transition = cfg.get('transition', 'fade') + self.trans_dur = cfg.get('transition_duration', 0.5) + self.bgm_volume = cfg.get('bgm_volume', 0.08) + self.burn_subs = cfg.get('burn_subtitles', True) + + def _check_ffmpeg(self) -> bool: + try: + r = subprocess.run( + [self.ffmpeg_path, '-version'], + capture_output=True, timeout=5, + ) + return r.returncode == 0 + except Exception: + return False + + def _run_ffmpeg(self, args: list, quiet: bool = True) -> bool: + cmd = [self.ffmpeg_path, '-y'] + if quiet: + cmd += ['-loglevel', 'error'] + cmd += args + result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) + if result.returncode != 0: + logger.error(f"ffmpeg 오류: {result.stderr[-400:]}") + return result.returncode == 0 + + def _get_audio_duration(self, wav_path: str) -> float: + try: + result = subprocess.run( + [self.ffprobe_path, '-v', 'quiet', '-print_format', 'json', + '-show_format', wav_path], + capture_output=True, text=True, timeout=10, + ) + data = json.loads(result.stdout) + return float(data['format']['duration']) + except Exception: + return 5.0 + + def _make_silent_wav(self, output_path: str, duration: float = 2.0) -> bool: + return self._run_ffmpeg([ + '-f', 'lavfi', '-i', f'anullsrc=r=24000:cl=mono', + '-t', str(duration), output_path, + ]) + + def _make_blank_slide(self, output_path: str) -> bool: + """단색(어두운) 빈 슬라이드 PNG 생성""" + try: + from PIL import Image, ImageDraw + img = Image.new('RGB', (1080, 1920), (10, 10, 13)) + draw = ImageDraw.Draw(img) + draw.rectangle([60, 950, 1020, 954], fill=(200, 168, 78)) + img.save(output_path) + return True + except ImportError: + # Pillow 없으면 ffmpeg lavfi로 단색 이미지 생성 + return self._run_ffmpeg([ + '-f', 'lavfi', '-i', 'color=c=black:s=1080x1920:r=1', + '-frames:v', '1', output_path, + ]) + + def _tts_gtts(self, text: str, output_path: str) -> bool: + try: + from gtts import gTTS + mp3_path = str(output_path).replace('.wav', '_tmp.mp3') + tts = gTTS(text=text, lang='ko', slow=False) + tts.save(mp3_path) + ok = self._run_ffmpeg(['-i', mp3_path, '-ar', '24000', output_path]) + Path(mp3_path).unlink(missing_ok=True) + return ok and Path(output_path).exists() + except Exception as e: + logger.warning(f"gTTS 실패: {e}") + return False + + def _make_clip(self, slide_png: str, audio_wav: str, output_mp4: str) -> float: + """슬라이드 PNG + 오디오 WAV → MP4 클립 (Ken Burns zoompan). 클립 길이(초) 반환.""" + duration = self._get_audio_duration(audio_wav) + 0.3 + ok = self._run_ffmpeg([ + '-loop', '1', '-i', slide_png, + '-i', audio_wav, + '-c:v', 'libx264', '-tune', 'stillimage', + '-c:a', 'aac', '-b:a', '192k', + '-pix_fmt', 'yuv420p', + '-vf', ( + 'scale=1080:1920,' + 'zoompan=z=\'min(zoom+0.0003,1.05)\':' + 'x=\'iw/2-(iw/zoom/2)\':' + 'y=\'ih/2-(ih/zoom/2)\':' + 'd=1:s=1080x1920:fps=30' + ), + '-shortest', + '-r', '30', + output_mp4, + ]) + return duration if ok else 0.0 + + def _concat_clips_xfade(self, clips: list, output_mp4: str) -> bool: + """여러 클립을 xfade 전환으로 결합""" + if len(clips) == 1: + shutil.copy2(clips[0]['mp4'], output_mp4) + return True + + n = len(clips) + inputs = [] + for c in clips: + inputs += ['-i', c['mp4']] + + filter_parts = [] + prev_v = '[0:v]' + prev_a = '[0:a]' + for i in range(1, n): + offset = sum(c['duration'] for c in clips[:i]) - self.trans_dur * i + out_v = f'[f{i}v]' if i < n - 1 else '[video]' + out_a = f'[f{i}a]' if i < n - 1 else '[audio]' + filter_parts.append( + f'{prev_v}[{i}:v]xfade=transition={self.transition}:' + f'duration={self.trans_dur}:offset={offset:.3f}{out_v}' + ) + filter_parts.append( + f'{prev_a}[{i}:a]acrossfade=d={self.trans_dur}{out_a}' + ) + prev_v = out_v + prev_a = out_a + + return self._run_ffmpeg( + inputs + [ + '-filter_complex', '; '.join(filter_parts), + '-map', '[video]', '-map', '[audio]', + '-c:v', 'libx264', '-c:a', 'aac', + '-pix_fmt', 'yuv420p', + output_mp4, + ] + ) + + def _mix_bgm(self, video_mp4: str, output_mp4: str) -> bool: + if not BGM_PATH.exists(): + logger.warning(f"BGM 파일 없음 ({BGM_PATH}) — BGM 없이 진행") + shutil.copy2(video_mp4, output_mp4) + return True + return self._run_ffmpeg([ + '-i', video_mp4, + '-i', str(BGM_PATH), + '-filter_complex', + f'[1:a]volume={self.bgm_volume}[bgm];[0:a][bgm]amix=inputs=2:duration=first[a]', + '-map', '0:v', '-map', '[a]', + '-c:v', 'copy', '-c:a', 'aac', + '-shortest', + output_mp4, + ]) + + def _burn_subtitles(self, video_mp4: str, srt_path: str, output_mp4: str) -> bool: + font_name = 'NanumGothic' + fonts_dir = ASSETS_DIR / 'fonts' + for fname in ['NotoSansKR-Regular.ttf', 'malgun.ttf']: + fp = fonts_dir / fname + if not fp.exists(): + fp = Path(f'C:/Windows/Fonts/{fname}') + if fp.exists(): + font_name = fp.stem + break + style = ( + f'FontName={font_name},' + 'FontSize=22,' + 'PrimaryColour=&H00FFFFFF,' + 'OutlineColour=&H80000000,' + 'BorderStyle=4,' + 'BackColour=&H80000000,' + 'Outline=0,Shadow=0,' + 'MarginV=120,' + 'Alignment=2,' + 'Bold=1' + ) + srt_esc = str(srt_path).replace('\\', '/').replace(':', '\\:') + return self._run_ffmpeg([ + '-i', video_mp4, + '-vf', f'subtitles={srt_esc}:force_style=\'{style}\'', + '-c:v', 'libx264', '-c:a', 'copy', + output_mp4, + ]) + + def _build_srt(self, scenes: list, clips: list) -> str: + lines = [] + t = 0.0 + for i, (scene, clip) in enumerate(zip(scenes, clips), 1): + text = scene.get('text', '') + if not text: + t += clip['duration'] - self.trans_dur + continue + end = t + clip['duration'] + mid = len(text) // 2 + if len(text) > 30: + space = text.rfind(' ', 0, mid) + if space > 0: + text = text[:space] + '\n' + text[space + 1:] + lines += [ + str(i), + f'{self._sec_to_srt(t)} --> {self._sec_to_srt(end)}', + text, + '', + ] + t += clip['duration'] - self.trans_dur + return '\n'.join(lines) + + @staticmethod + def _sec_to_srt(s: float) -> str: + h, rem = divmod(int(s), 3600) + m, sec = divmod(rem, 60) + ms = int((s - int(s)) * 1000) + return f'{h:02d}:{m:02d}:{sec:02d},{ms:03d}' + + def generate(self, scenes: list, output_path: str, **kwargs) -> str: + """ + scenes 리스트로 쇼츠 MP4 생성. + + kwargs: + article (dict): 원본 article 데이터 (슬라이드 합성에 사용) + tts_engine: BaseTTS 인스턴스 (없으면 GTTSEngine 사용) + """ + if not self._check_ffmpeg(): + logger.error("ffmpeg 없음. PATH 또는 FFMPEG_PATH 환경변수 확인") + return '' + + if not scenes: + logger.warning("scenes 비어 있음 — 영상 생성 불가") + return '' + + logger.info(f"FFmpegSlidesEngine 시작: {len(scenes)}개 씬 → {output_path}") + + tts_engine = kwargs.get('tts_engine', None) + + with tempfile.TemporaryDirectory() as tmp: + tmp_dir = Path(tmp) + clips = [] + + for idx, scene in enumerate(scenes): + scene_key = scene.get('type', f'scene{idx}') + + # ── 슬라이드 준비 ────────────────────── + slide_path = scene.get('slide_path', '') + if not slide_path or not Path(slide_path).exists(): + # shorts_converter의 슬라이드 합성 함수 재사용 시도 + slide_path = str(tmp_dir / f'slide_{idx}.png') + article = kwargs.get('article', {}) + composed = self._compose_scene_slide( + scene, idx, article, tmp_dir + ) + if composed: + slide_path = composed + else: + self._make_blank_slide(slide_path) + + # ── 오디오 준비 ──────────────────────── + audio_path = scene.get('audio_path', '') + if not audio_path or not Path(audio_path).exists(): + audio_path = str(tmp_dir / f'tts_{idx}.wav') + text = scene.get('text', '') + ok = False + if tts_engine and text: + try: + ok = tts_engine.synthesize(text, audio_path) + except Exception as e: + logger.warning(f"TTS 엔진 실패: {e}") + if not ok and text: + ok = self._tts_gtts(text, audio_path) + if not ok: + self._make_silent_wav(audio_path) + + # ── 클립 생성 ────────────────────────── + clip_path = str(tmp_dir / f'clip_{idx}.mp4') + dur = self._make_clip(slide_path, audio_path, clip_path) + if dur > 0: + clips.append({'mp4': clip_path, 'duration': dur}) + else: + logger.warning(f"씬 {idx} ({scene_key}) 클립 생성 실패 — 건너뜀") + + if not clips: + logger.error("생성된 클립 없음") + return '' + + # ── 클립 결합 ────────────────────────────── + merged = str(tmp_dir / 'merged.mp4') + if not self._concat_clips_xfade(clips, merged): + logger.error("클립 결합 실패") + return '' + + # ── BGM 믹스 ─────────────────────────────── + with_bgm = str(tmp_dir / 'with_bgm.mp4') + self._mix_bgm(merged, with_bgm) + source_for_srt = with_bgm if Path(with_bgm).exists() else merged + + # ── 자막 burn-in ─────────────────────────── + if self.burn_subs: + srt_content = self._build_srt(scenes, clips) + srt_path = str(tmp_dir / 'subtitles.srt') + Path(srt_path).write_text(srt_content, encoding='utf-8-sig') + + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + if not self._burn_subtitles(source_for_srt, srt_path, output_path): + logger.warning("자막 burn-in 실패 — 자막 없는 버전으로 저장") + shutil.copy2(source_for_srt, output_path) + else: + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source_for_srt, output_path) + + if Path(output_path).exists(): + logger.info(f"FFmpegSlidesEngine 완료: {output_path}") + return output_path + else: + logger.error(f"최종 파일 없음: {output_path}") + return '' + + def _compose_scene_slide(self, scene: dict, idx: int, + article: dict, tmp_dir: Path) -> Optional[str]: + """ + shorts_converter의 슬라이드 합성 함수를 재사용해 씬별 슬라이드 생성. + 임포트 실패 시 None 반환 (blank slide 폴백). + """ + try: + from bots.converters.shorts_converter import ( + compose_intro_slide, + compose_headline_slide, + compose_point_slide, + compose_outro_slide, + compose_data_slide, + _set_tmp_dir, + _load_template, + ) + _set_tmp_dir(tmp_dir) + cfg = _load_template() + scene_type = scene.get('type', '') + out_path = str(tmp_dir / f'slide_{idx}.png') + + if scene_type == 'intro': + return compose_intro_slide(cfg) + elif scene_type == 'headline': + return compose_headline_slide(article, cfg) + elif scene_type in ('point', 'point1', 'point2', 'point3'): + num = int(scene_type[-1]) if scene_type[-1].isdigit() else 1 + return compose_point_slide(scene.get('text', ''), num, article, cfg) + elif scene_type == 'data': + return compose_data_slide(article, cfg) + elif scene_type == 'outro': + return compose_outro_slide(cfg) + else: + # 알 수 없는 타입 → 헤드라인 슬라이드로 대체 + return compose_headline_slide(article, cfg) + except ImportError as e: + logger.warning(f"shorts_converter 임포트 실패: {e}") + return None + except Exception as e: + logger.warning(f"슬라이드 합성 실패 (씬 {idx}): {e}") + return None + + +# ─── SeedanceEngine ──────────────────────────────────── + +class SeedanceEngine(VideoEngine): + """ + Seedance 2.0 API를 사용한 AI 영상 생성 엔진. + API 키 없거나 실패 시 FFmpegSlidesEngine으로 자동 폴백. + """ + + def __init__(self, cfg: dict): + self.api_url = cfg.get('api_url', 'https://api.seedance2.ai/v1/generate') + self.api_key = os.getenv(cfg.get('api_key_env', 'SEEDANCE_API_KEY'), '') + self.resolution = cfg.get('resolution', '1080x1920') + self.duration = cfg.get('duration', '10s') + self.audio = cfg.get('audio', True) + self._fallback_cfg = cfg + + def _fallback(self, scenes: list, output_path: str, **kwargs) -> str: + logger.info("SeedanceEngine → FFmpegSlidesEngine 폴백") + return FFmpegSlidesEngine(self._fallback_cfg).generate( + scenes, output_path, **kwargs + ) + + def _download_file(self, url: str, dest: str, timeout: int = 120) -> bool: + try: + import requests as req + resp = req.get(url, timeout=timeout, stream=True) + resp.raise_for_status() + with open(dest, 'wb') as f: + for chunk in resp.iter_content(chunk_size=8192): + f.write(chunk) + return True + except Exception as e: + logger.error(f"파일 다운로드 실패 ({url}): {e}") + return False + + def _concat_clips_ffmpeg(self, clip_paths: list, output_path: str) -> bool: + """ffmpeg concat demuxer로 클립 결합 (인트로 2초 + 씬 + 아웃트로 3초)""" + if not clip_paths: + return False + ffmpeg = os.getenv('FFMPEG_PATH', 'ffmpeg') + with tempfile.TemporaryDirectory() as tmp: + list_file = str(Path(tmp) / 'clips.txt') + with open(list_file, 'w', encoding='utf-8') as f: + for p in clip_paths: + f.write(f"file '{p}'\n") + result = subprocess.run( + [ffmpeg, '-y', '-loglevel', 'error', + '-f', 'concat', '-safe', '0', + '-i', list_file, + '-c', 'copy', output_path], + capture_output=True, timeout=300, + ) + return result.returncode == 0 + + def _generate_scene_clip(self, scene: dict, output_path: str) -> bool: + """단일 씬에 대해 Seedance API 호출 → 클립 다운로드""" + try: + import requests as req + prompt = scene.get('image_prompt') or scene.get('text', '') + if not prompt: + return False + + payload = { + 'prompt': prompt, + 'resolution': self.resolution, + 'duration': self.duration, + 'audio': self.audio, + } + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + } + logger.info(f"Seedance API 호출: {prompt[:60]}...") + resp = req.post(self.api_url, json=payload, headers=headers, timeout=120) + resp.raise_for_status() + + data = resp.json() + video_url = data.get('video_url') or data.get('url', '') + if not video_url: + logger.error(f"Seedance 응답에 video_url 없음: {data}") + return False + + return self._download_file(video_url, output_path) + except Exception as e: + logger.error(f"Seedance API 오류: {e}") + return False + + def generate(self, scenes: list, output_path: str, **kwargs) -> str: + if not self.api_key: + logger.warning("SEEDANCE_API_KEY 없음 — FFmpegSlidesEngine으로 폴백") + return self._fallback(scenes, output_path, **kwargs) + + if not scenes: + logger.warning("scenes 비어 있음") + return '' + + logger.info(f"SeedanceEngine 시작: {len(scenes)}개 씬") + + ffmpeg = os.getenv('FFMPEG_PATH', 'ffmpeg') + + with tempfile.TemporaryDirectory() as tmp: + tmp_dir = Path(tmp) + clip_paths = [] + + # 인트로 클립 (2초 단색) + intro_path = str(tmp_dir / 'intro.mp4') + subprocess.run( + [ffmpeg, '-y', '-loglevel', 'error', + '-f', 'lavfi', '-i', 'color=c=black:s=1080x1920:r=30', + '-t', '2', '-c:v', 'libx264', '-pix_fmt', 'yuv420p', + intro_path], + capture_output=True, timeout=30, + ) + if Path(intro_path).exists(): + clip_paths.append(intro_path) + + # 씬별 클립 생성 + success_count = 0 + for idx, scene in enumerate(scenes): + clip_path = str(tmp_dir / f'scene_{idx}.mp4') + if self._generate_scene_clip(scene, clip_path): + clip_paths.append(clip_path) + success_count += 1 + else: + logger.warning(f"씬 {idx} Seedance 실패 — 폴백으로 전환") + return self._fallback(scenes, output_path, **kwargs) + + if success_count == 0: + logger.warning("모든 씬 실패 — FFmpegSlidesEngine으로 폴백") + return self._fallback(scenes, output_path, **kwargs) + + # 아웃트로 클립 (3초 단색) + outro_path = str(tmp_dir / 'outro.mp4') + subprocess.run( + [ffmpeg, '-y', '-loglevel', 'error', + '-f', 'lavfi', '-i', 'color=c=black:s=1080x1920:r=30', + '-t', '3', '-c:v', 'libx264', '-pix_fmt', 'yuv420p', + outro_path], + capture_output=True, timeout=30, + ) + if Path(outro_path).exists(): + clip_paths.append(outro_path) + + # 클립 결합 + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + if not self._concat_clips_ffmpeg(clip_paths, output_path): + logger.error("SeedanceEngine 클립 결합 실패") + return self._fallback(scenes, output_path, **kwargs) + + if Path(output_path).exists(): + logger.info(f"SeedanceEngine 완료: {output_path}") + return output_path + return self._fallback(scenes, output_path, **kwargs) + + +# ─── SoraEngine ──────────────────────────────────────── + +class SoraEngine(VideoEngine): + """ + OpenAI Sora 영상 생성 엔진. + 현재 API 공개 접근 불가 — ffmpeg_slides로 폴백. + """ + + def __init__(self, cfg: dict): + self.cfg = cfg + + def generate(self, scenes: list, output_path: str, **kwargs) -> str: + logger.warning("Sora API 미지원. ffmpeg_slides로 폴백.") + return FFmpegSlidesEngine(self.cfg).generate(scenes, output_path, **kwargs) + + +# ─── RunwayEngine ────────────────────────────────────── + +class RunwayEngine(VideoEngine): + """ + Runway Gen-3 API를 사용한 AI 영상 생성 엔진. + API 키 없거나 실패 시 FFmpegSlidesEngine으로 자동 폴백. + """ + + def __init__(self, cfg: dict): + self.cfg = cfg + self.api_key = os.getenv(cfg.get('api_key_env', 'RUNWAY_API_KEY'), '') + self.api_url = cfg.get('api_url', 'https://api.runwayml.com/v1/image_to_video') + self.model = cfg.get('model', 'gen3a_turbo') + self.duration = cfg.get('duration', 10) + self.ratio = cfg.get('ratio', '768:1344') + + def _fallback(self, scenes: list, output_path: str, **kwargs) -> str: + logger.info("RunwayEngine → FFmpegSlidesEngine 폴백") + return FFmpegSlidesEngine(self.cfg).generate(scenes, output_path, **kwargs) + + def _generate_scene_clip(self, scene: dict, output_path: str) -> bool: + """단일 씬에 대해 Runway API 호출 → 클립 다운로드""" + try: + import requests as req + prompt = scene.get('image_prompt') or scene.get('text', '') + if not prompt: + return False + + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + 'X-Runway-Version': '2024-11-06', + } + payload = { + 'model': self.model, + 'promptText': prompt, + 'duration': self.duration, + 'ratio': self.ratio, + } + logger.info(f"Runway API 호출: {prompt[:60]}...") + resp = req.post(self.api_url, json=payload, headers=headers, timeout=30) + resp.raise_for_status() + + data = resp.json() + task_id = data.get('id', '') + if not task_id: + logger.error(f"Runway 태스크 ID 없음: {data}") + return False + + # 폴링: 태스크 완료 대기 + poll_url = f'https://api.runwayml.com/v1/tasks/{task_id}' + import time + for _ in range(60): + time.sleep(10) + poll = req.get(poll_url, headers=headers, timeout=30) + poll.raise_for_status() + status_data = poll.json() + status = status_data.get('status', '') + if status == 'SUCCEEDED': + video_url = (status_data.get('output') or [''])[0] + if not video_url: + logger.error("Runway 완료됐으나 video_url 없음") + return False + return self._download_file(video_url, output_path) + elif status in ('FAILED', 'CANCELLED'): + logger.error(f"Runway 태스크 실패: {status_data}") + return False + logger.error("Runway 태스크 타임아웃 (10분)") + return False + except Exception as e: + logger.error(f"Runway API 오류: {e}") + return False + + def _download_file(self, url: str, dest: str, timeout: int = 120) -> bool: + try: + import requests as req + resp = req.get(url, timeout=timeout, stream=True) + resp.raise_for_status() + with open(dest, 'wb') as f: + for chunk in resp.iter_content(chunk_size=8192): + f.write(chunk) + return True + except Exception as e: + logger.error(f"파일 다운로드 실패 ({url}): {e}") + return False + + def generate(self, scenes: list, output_path: str, **kwargs) -> str: + if not self.api_key: + logger.warning("RUNWAY_API_KEY 없음 — FFmpegSlidesEngine으로 폴백") + return self._fallback(scenes, output_path, **kwargs) + + if not scenes: + logger.warning("scenes 비어 있음") + return '' + + logger.info(f"RunwayEngine 시작: {len(scenes)}개 씬") + + ffmpeg = os.getenv('FFMPEG_PATH', 'ffmpeg') + + with tempfile.TemporaryDirectory() as tmp: + tmp_dir = Path(tmp) + clip_paths = [] + + for idx, scene in enumerate(scenes): + clip_path = str(tmp_dir / f'scene_{idx}.mp4') + if self._generate_scene_clip(scene, clip_path): + clip_paths.append(clip_path) + else: + logger.warning(f"씬 {idx} Runway 실패 — FFmpegSlidesEngine 폴백") + return self._fallback(scenes, output_path, **kwargs) + + if not clip_paths: + return self._fallback(scenes, output_path, **kwargs) + + # concat + list_file = str(tmp_dir / 'clips.txt') + with open(list_file, 'w', encoding='utf-8') as f: + for p in clip_paths: + f.write(f"file '{p}'\n") + + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + result = subprocess.run( + [ffmpeg, '-y', '-loglevel', 'error', + '-f', 'concat', '-safe', '0', + '-i', list_file, '-c', 'copy', output_path], + capture_output=True, timeout=300, + ) + if result.returncode != 0: + logger.error("RunwayEngine 클립 결합 실패") + return self._fallback(scenes, output_path, **kwargs) + + if Path(output_path).exists(): + logger.info(f"RunwayEngine 완료: {output_path}") + return output_path + return self._fallback(scenes, output_path, **kwargs) + + +# ─── VeoEngine ───────────────────────────────────────── + +class VeoEngine(VideoEngine): + """ + Google Veo 3.1 영상 생성 엔진. + 현재 API 공개 접근 불가 — ffmpeg_slides로 폴백. + """ + + def __init__(self, cfg: dict): + self.cfg = cfg + + def generate(self, scenes: list, output_path: str, **kwargs) -> str: + logger.warning("Veo API 미지원. ffmpeg_slides로 폴백.") + return FFmpegSlidesEngine(self.cfg).generate(scenes, output_path, **kwargs) + + +# ─── 팩토리 함수 ─────────────────────────────────────── + +def get_engine(video_cfg: dict) -> VideoEngine: + """ + engine.json video_generation 설정에서 엔진 인스턴스 반환. + + 사용: + cfg = {'provider': 'ffmpeg_slides', 'options': {...}} + engine = get_engine(cfg) + mp4 = engine.generate(scenes, '/path/to/output.mp4') + """ + provider = video_cfg.get('provider', 'ffmpeg_slides') + opts = video_cfg.get('options', {}).get(provider, {}) + + engine_map = { + 'ffmpeg_slides': FFmpegSlidesEngine, + 'seedance': SeedanceEngine, + 'sora': SoraEngine, + 'runway': RunwayEngine, + 'veo': VeoEngine, + } + cls = engine_map.get(provider, FFmpegSlidesEngine) + logger.info(f"VideoEngine 선택: {provider} ({cls.__name__})") + return cls(opts) diff --git a/bots/engine_loader.py b/bots/engine_loader.py new file mode 100644 index 0000000..4343f04 --- /dev/null +++ b/bots/engine_loader.py @@ -0,0 +1,521 @@ +""" +엔진 로더 (bots/engine_loader.py) +역할: config/engine.json을 읽어 현재 설정된 provider에 맞는 구현체를 반환 +설계서: blog-engine-final-masterplan-v3.txt + +사용: + loader = EngineLoader() + writer = loader.get_writer() + result = writer.write("AI 관련 기사 써줘") + tts = loader.get_tts() + tts.synthesize("안녕하세요", "/tmp/out.wav") +""" +import json +import logging +import os +import subprocess +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Any, Optional + +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = Path(__file__).parent.parent +CONFIG_PATH = BASE_DIR / 'config' / 'engine.json' +LOG_DIR = BASE_DIR / 'logs' +LOG_DIR.mkdir(exist_ok=True) + +logger = logging.getLogger(__name__) +if not logger.handlers: + handler = logging.FileHandler(LOG_DIR / 'engine_loader.log', encoding='utf-8') + handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')) + logger.addHandler(handler) + logger.addHandler(logging.StreamHandler()) + logger.setLevel(logging.INFO) + + +# ─── 기본 추상 클래스 ────────────────────────────────── + +class BaseWriter(ABC): + @abstractmethod + def write(self, prompt: str, system: str = '') -> str: + """글쓰기 요청. prompt에 대한 결과 문자열 반환.""" + + +class BaseTTS(ABC): + @abstractmethod + def synthesize(self, text: str, output_path: str, + lang: str = 'ko', speed: float = 1.05) -> bool: + """TTS 합성. 성공 시 True 반환.""" + + +class BaseImageGenerator(ABC): + @abstractmethod + def generate(self, prompt: str, output_path: str, + size: str = '1024x1792') -> bool: + """이미지 생성. 성공 시 True 반환.""" + + +# VideoEngine은 video_engine.py에 정의됨 +# BaseVideoGenerator 타입 힌트 호환용 +BaseVideoGenerator = object + + +# ─── Writer 구현체 ────────────────────────────────────── + +class ClaudeWriter(BaseWriter): + """Anthropic Claude API를 사용하는 글쓰기 엔진""" + + def __init__(self, cfg: dict): + self.api_key = os.getenv(cfg.get('api_key_env', 'ANTHROPIC_API_KEY'), '') + self.model = cfg.get('model', 'claude-opus-4-5') + self.max_tokens = cfg.get('max_tokens', 4096) + self.temperature = cfg.get('temperature', 0.7) + + def write(self, prompt: str, system: str = '') -> str: + if not self.api_key: + logger.warning("ANTHROPIC_API_KEY 없음 — ClaudeWriter 비활성화") + return '' + try: + import anthropic + client = anthropic.Anthropic(api_key=self.api_key) + kwargs: dict = { + 'model': self.model, + 'max_tokens': self.max_tokens, + 'messages': [{'role': 'user', 'content': prompt}], + } + if system: + kwargs['system'] = system + message = client.messages.create(**kwargs) + return message.content[0].text + except Exception as e: + logger.error(f"ClaudeWriter 오류: {e}") + return '' + + +class OpenClawWriter(BaseWriter): + """OpenClaw CLI를 subprocess로 호출하는 글쓰기 엔진""" + + def __init__(self, cfg: dict): + self.agent_name = cfg.get('agent_name', 'blog-writer') + self.timeout = cfg.get('timeout', 120) + + def write(self, prompt: str, system: str = '') -> str: + try: + cmd = ['openclaw', 'run', self.agent_name, '--prompt', prompt] + if system: + cmd += ['--system', system] + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=self.timeout, + encoding='utf-8', + ) + if result.returncode != 0: + logger.error(f"OpenClawWriter 오류: {result.stderr[:300]}") + return '' + return result.stdout.strip() + except subprocess.TimeoutExpired: + logger.error(f"OpenClawWriter 타임아웃 ({self.timeout}초)") + return '' + except FileNotFoundError: + logger.warning("openclaw CLI 없음 — OpenClawWriter 비활성화") + return '' + except Exception as e: + logger.error(f"OpenClawWriter 오류: {e}") + return '' + + +class GeminiWriter(BaseWriter): + """Google Gemini API를 사용하는 글쓰기 엔진""" + + def __init__(self, cfg: dict): + self.api_key = os.getenv(cfg.get('api_key_env', 'GEMINI_API_KEY'), '') + self.model = cfg.get('model', 'gemini-2.0-flash') + self.max_tokens = cfg.get('max_tokens', 4096) + self.temperature = cfg.get('temperature', 0.7) + + def write(self, prompt: str, system: str = '') -> str: + if not self.api_key: + logger.warning("GEMINI_API_KEY 없음 — GeminiWriter 비활성화") + return '' + try: + import google.generativeai as genai # type: ignore + genai.configure(api_key=self.api_key) + model = genai.GenerativeModel( + model_name=self.model, + generation_config={ + 'max_output_tokens': self.max_tokens, + 'temperature': self.temperature, + }, + system_instruction=system if system else None, + ) + response = model.generate_content(prompt) + return response.text + except ImportError: + logger.warning("google-generativeai 미설치 — GeminiWriter 비활성화") + return '' + except Exception as e: + logger.error(f"GeminiWriter 오류: {e}") + return '' + + +# ─── TTS 구현체 ───────────────────────────────────────── + +class GoogleCloudTTS(BaseTTS): + """Google Cloud TTS REST API (API Key 방식)""" + + def __init__(self, cfg: dict): + self.api_key = os.getenv(cfg.get('api_key_env', 'GOOGLE_TTS_API_KEY'), '') + self.voice = cfg.get('voice', 'ko-KR-Wavenet-A') + self.default_speed = cfg.get('speaking_rate', 1.05) + self.pitch = cfg.get('pitch', 0) + + def synthesize(self, text: str, output_path: str, + lang: str = 'ko', speed: float = 0.0) -> bool: + if not self.api_key: + logger.warning("GOOGLE_TTS_API_KEY 없음 — GoogleCloudTTS 비활성화") + return False + import base64 + try: + import requests as req + speaking_rate = speed if speed > 0 else self.default_speed + voice_name = self.voice if lang == 'ko' else 'en-US-Wavenet-D' + language_code = 'ko-KR' if lang == 'ko' else 'en-US' + url = ( + f'https://texttospeech.googleapis.com/v1/text:synthesize' + f'?key={self.api_key}' + ) + payload = { + 'input': {'text': text}, + 'voice': {'languageCode': language_code, 'name': voice_name}, + 'audioConfig': { + 'audioEncoding': 'LINEAR16', + 'speakingRate': speaking_rate, + 'pitch': self.pitch, + }, + } + resp = req.post(url, json=payload, timeout=30) + resp.raise_for_status() + audio_b64 = resp.json().get('audioContent', '') + if audio_b64: + Path(output_path).write_bytes(base64.b64decode(audio_b64)) + return True + except Exception as e: + logger.warning(f"GoogleCloudTTS 실패: {e}") + return False + + +class OpenAITTS(BaseTTS): + """OpenAI TTS API (tts-1-hd)""" + + def __init__(self, cfg: dict): + self.api_key = os.getenv(cfg.get('api_key_env', 'OPENAI_API_KEY'), '') + self.model = cfg.get('model', 'tts-1-hd') + self.voice = cfg.get('voice', 'alloy') + self.default_speed = cfg.get('speed', 1.0) + + def synthesize(self, text: str, output_path: str, + lang: str = 'ko', speed: float = 0.0) -> bool: + if not self.api_key: + logger.warning("OPENAI_API_KEY 없음 — OpenAITTS 비활성화") + return False + try: + from openai import OpenAI + client = OpenAI(api_key=self.api_key) + speak_speed = speed if speed > 0 else self.default_speed + response = client.audio.speech.create( + model=self.model, + voice=self.voice, + input=text, + speed=speak_speed, + response_format='wav', + ) + response.stream_to_file(output_path) + return Path(output_path).exists() + except ImportError: + logger.warning("openai 미설치 — OpenAITTS 비활성화") + return False + except Exception as e: + logger.error(f"OpenAITTS 실패: {e}") + return False + + +class ElevenLabsTTS(BaseTTS): + """ElevenLabs REST API TTS""" + + def __init__(self, cfg: dict): + self.api_key = os.getenv(cfg.get('api_key_env', 'ELEVENLABS_API_KEY'), '') + self.model = cfg.get('model', 'eleven_multilingual_v2') + self.voice_id = cfg.get('voice_id', 'pNInz6obpgDQGcFmaJgB') + self.stability = cfg.get('stability', 0.5) + self.similarity_boost = cfg.get('similarity_boost', 0.75) + + def synthesize(self, text: str, output_path: str, + lang: str = 'ko', speed: float = 0.0) -> bool: + if not self.api_key: + logger.warning("ELEVENLABS_API_KEY 없음 — ElevenLabsTTS 비활성화") + return False + try: + import requests as req + url = ( + f'https://api.elevenlabs.io/v1/text-to-speech/' + f'{self.voice_id}' + ) + headers = { + 'xi-api-key': self.api_key, + 'Content-Type': 'application/json', + 'Accept': 'audio/mpeg', + } + payload = { + 'text': text, + 'model_id': self.model, + 'voice_settings': { + 'stability': self.stability, + 'similarity_boost': self.similarity_boost, + }, + } + resp = req.post(url, json=payload, headers=headers, timeout=60) + resp.raise_for_status() + # mp3 응답 → 파일 저장 (wav 확장자라도 mp3 데이터 저장 후 ffmpeg 변환) + mp3_path = str(output_path).replace('.wav', '_tmp.mp3') + Path(mp3_path).write_bytes(resp.content) + # mp3 → wav 변환 (ffmpeg 사용) + ffmpeg = os.getenv('FFMPEG_PATH', 'ffmpeg') + result = subprocess.run( + [ffmpeg, '-y', '-loglevel', 'error', '-i', mp3_path, + '-ar', '24000', output_path], + capture_output=True, timeout=60, + ) + Path(mp3_path).unlink(missing_ok=True) + return Path(output_path).exists() and result.returncode == 0 + except Exception as e: + logger.error(f"ElevenLabsTTS 실패: {e}") + return False + + +class GTTSEngine(BaseTTS): + """gTTS 무료 TTS 엔진""" + + def __init__(self, cfg: dict): + self.default_lang = cfg.get('lang', 'ko') + self.slow = cfg.get('slow', False) + + def synthesize(self, text: str, output_path: str, + lang: str = 'ko', speed: float = 0.0) -> bool: + try: + from gtts import gTTS + use_lang = lang if lang else self.default_lang + mp3_path = str(output_path).replace('.wav', '_tmp.mp3') + tts = gTTS(text=text, lang=use_lang, slow=self.slow) + tts.save(mp3_path) + # mp3 → wav 변환 (ffmpeg 사용) + ffmpeg = os.getenv('FFMPEG_PATH', 'ffmpeg') + result = subprocess.run( + [ffmpeg, '-y', '-loglevel', 'error', '-i', mp3_path, + '-ar', '24000', output_path], + capture_output=True, timeout=60, + ) + Path(mp3_path).unlink(missing_ok=True) + return Path(output_path).exists() and result.returncode == 0 + except ImportError: + logger.warning("gTTS 미설치 — GTTSEngine 비활성화") + return False + except Exception as e: + logger.warning(f"GTTSEngine 실패: {e}") + return False + + +# ─── ImageGenerator 구현체 ───────────────────────────── + +class DALLEGenerator(BaseImageGenerator): + """OpenAI DALL-E 3 이미지 생성 엔진""" + + def __init__(self, cfg: dict): + self.api_key = os.getenv(cfg.get('api_key_env', 'OPENAI_API_KEY'), '') + self.model = cfg.get('model', 'dall-e-3') + self.default_size = cfg.get('size', '1024x1792') + self.quality = cfg.get('quality', 'standard') + + def generate(self, prompt: str, output_path: str, + size: str = '') -> bool: + if not self.api_key: + logger.warning("OPENAI_API_KEY 없음 — DALLEGenerator 비활성화") + return False + try: + from openai import OpenAI + import requests as req + import io + from PIL import Image + + use_size = size if size else self.default_size + client = OpenAI(api_key=self.api_key) + full_prompt = prompt + ' No text, no letters, no numbers, no watermarks.' + response = client.images.generate( + model=self.model, + prompt=full_prompt, + size=use_size, + quality=self.quality, + n=1, + ) + img_url = response.data[0].url + img_bytes = req.get(img_url, timeout=30).content + img = Image.open(io.BytesIO(img_bytes)).convert('RGB') + img.save(output_path) + logger.info(f"DALL-E 이미지 생성 완료: {output_path}") + return True + except ImportError as e: + logger.warning(f"DALLEGenerator 의존성 없음: {e}") + return False + except Exception as e: + logger.error(f"DALLEGenerator 실패: {e}") + return False + + +class ExternalGenerator(BaseImageGenerator): + """수동 이미지 제공 (자동 생성 없음)""" + + def __init__(self, cfg: dict): + pass + + def generate(self, prompt: str, output_path: str, + size: str = '') -> bool: + logger.info(f"ExternalGenerator: 수동 이미지 필요 — 프롬프트: {prompt[:60]}") + return False + + +# ─── EngineLoader ─────────────────────────────────────── + +class EngineLoader: + """ + config/engine.json을 읽어 현재 설정된 provider에 맞는 구현체를 반환하는 + 중앙 팩토리 클래스. + + 사용 예: + loader = EngineLoader() + writer = loader.get_writer() + text = writer.write("오늘의 AI 뉴스 정리해줘") + """ + + _DEFAULT_CONFIG = { + 'writing': {'provider': 'claude', 'options': {'claude': {}}}, + 'tts': {'provider': 'gtts', 'options': {'gtts': {}}}, + 'image_generation': {'provider': 'external', 'options': {'external': {}}}, + 'video_generation': {'provider': 'ffmpeg_slides', 'options': {'ffmpeg_slides': {}}}, + 'publishing': {}, + 'quality_gates': {'gate1_research_min_score': 60}, + } + + def __init__(self, config_path: Optional[Path] = None): + self._config_path = config_path or CONFIG_PATH + self._config = self._load_config() + + def _load_config(self) -> dict: + if self._config_path.exists(): + try: + return json.loads(self._config_path.read_text(encoding='utf-8')) + except Exception as e: + logger.error(f"engine.json 로드 실패: {e} — 기본값 사용") + else: + logger.warning(f"engine.json 없음 ({self._config_path}) — 기본값으로 gtts + ffmpeg_slides 사용") + return dict(self._DEFAULT_CONFIG) + + def get_config(self, *keys) -> Any: + """ + engine.json 값 접근. + 예: loader.get_config('writing', 'provider') + loader.get_config('quality_gates', 'gate1_research_min_score') + """ + val = self._config + for key in keys: + if isinstance(val, dict): + val = val.get(key) + else: + return None + return val + + def update_provider(self, category: str, provider: str) -> None: + """ + 런타임 provider 변경 (engine.json 파일은 수정하지 않음). + 예: loader.update_provider('tts', 'openai') + """ + if category in self._config: + self._config[category]['provider'] = provider + logger.info(f"런타임 provider 변경: {category} → {provider}") + else: + logger.warning(f"update_provider: 알 수 없는 카테고리 '{category}'") + + def get_writer(self) -> BaseWriter: + """현재 설정된 writing provider에 맞는 BaseWriter 구현체 반환""" + writing_cfg = self._config.get('writing', {}) + provider = writing_cfg.get('provider', 'claude') + options = writing_cfg.get('options', {}).get(provider, {}) + + writers = { + 'claude': ClaudeWriter, + 'openclaw': OpenClawWriter, + 'gemini': GeminiWriter, + } + cls = writers.get(provider, ClaudeWriter) + logger.info(f"Writer 로드: {provider} ({cls.__name__})") + return cls(options) + + def get_tts(self) -> BaseTTS: + """현재 설정된 tts provider에 맞는 BaseTTS 구현체 반환""" + tts_cfg = self._config.get('tts', {}) + provider = tts_cfg.get('provider', 'gtts') + options = tts_cfg.get('options', {}).get(provider, {}) + + tts_engines = { + 'google_cloud': GoogleCloudTTS, + 'openai': OpenAITTS, + 'elevenlabs': ElevenLabsTTS, + 'gtts': GTTSEngine, + } + cls = tts_engines.get(provider, GTTSEngine) + logger.info(f"TTS 로드: {provider} ({cls.__name__})") + return cls(options) + + def get_image_generator(self) -> BaseImageGenerator: + """현재 설정된 image_generation provider에 맞는 구현체 반환""" + img_cfg = self._config.get('image_generation', {}) + provider = img_cfg.get('provider', 'external') + options = img_cfg.get('options', {}).get(provider, {}) + + generators = { + 'dalle': DALLEGenerator, + 'external': ExternalGenerator, + } + cls = generators.get(provider, ExternalGenerator) + logger.info(f"ImageGenerator 로드: {provider} ({cls.__name__})") + return cls(options) + + def get_video_generator(self): + """현재 설정된 video_generation provider에 맞는 VideoEngine 구현체 반환""" + from bots.converters import video_engine + video_cfg = self._config.get('video_generation', { + 'provider': 'ffmpeg_slides', + 'options': {'ffmpeg_slides': {}}, + }) + engine = video_engine.get_engine(video_cfg) + logger.info(f"VideoGenerator 로드: {video_cfg.get('provider', 'ffmpeg_slides')}") + return engine + + def get_publishers(self) -> list: + """ + 활성화된 publishing 채널 목록 반환. + 반환 형식: [{'name': str, 'enabled': bool, ...설정값}, ...] + """ + publishing_cfg = self._config.get('publishing', {}) + result = [] + for name, cfg in publishing_cfg.items(): + if isinstance(cfg, dict): + result.append({'name': name, **cfg}) + return result + + def get_enabled_publishers(self) -> list: + """enabled: true인 publishing 채널만 반환""" + return [p for p in self.get_publishers() if p.get('enabled', False)] diff --git a/bots/novel/__init__.py b/bots/novel/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bots/novel/novel_blog_converter.py b/bots/novel/novel_blog_converter.py new file mode 100644 index 0000000..89aef64 --- /dev/null +++ b/bots/novel/novel_blog_converter.py @@ -0,0 +1,341 @@ +""" +novel_blog_converter.py +소설 연재 파이프라인 — 에피소드 → Blogger-ready HTML 변환 모듈 +역할: 에피소드 dict + 소설 설정 → 장르별 테마 HTML 생성 +출력: data/novels/{novel_id}/episodes/ep{N:03d}_blog.html +""" +import json +import logging +import sys +from datetime import datetime, timezone +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = Path(__file__).parent.parent.parent +sys.path.insert(0, str(BASE_DIR / 'bots')) + +logger = logging.getLogger(__name__) + +BLOG_BASE_URL = 'https://the4thpath.com' + +# ─── 장르별 컬러 테마 ───────────────────────────────────────────────────────── + +GENRE_THEMES = { + 'sci-fi': { + 'bg': '#0a0f1e', + 'accent': '#00bcd4', + 'accent_dim': '#007c8c', + 'card_bg': '#0e1628', + 'text': '#cfe8ef', + 'meta': '#6fa8bc', + 'nav_bg': '#0c1220', + }, + 'thriller': { + 'bg': '#0a0a0d', + 'accent': '#bf3a3a', + 'accent_dim': '#8a2222', + 'card_bg': '#141418', + 'text': '#e8e0e0', + 'meta': '#a08080', + 'nav_bg': '#111115', + }, + 'fantasy': { + 'bg': '#0f0a1e', + 'accent': '#c8a84e', + 'accent_dim': '#8a7030', + 'card_bg': '#180f2e', + 'text': '#e8e0f0', + 'meta': '#9a8ab0', + 'nav_bg': '#130c22', + }, + 'romance': { + 'bg': '#ffffff', + 'accent': '#d85a30', + 'accent_dim': '#a04020', + 'card_bg': '#fff5f0', + 'text': '#2a1a14', + 'meta': '#8a5a4a', + 'nav_bg': '#fff0ea', + }, + 'default': { + 'bg': '#0a0a0d', + 'accent': '#c8a84e', + 'accent_dim': '#8a7030', + 'card_bg': '#141418', + 'text': '#e8e0d0', + 'meta': '#a09070', + 'nav_bg': '#111115', + }, +} + + +def _get_theme(genre: str) -> dict: + """장르 문자열에서 테마 결정 (부분 매칭 포함)""" + genre_lower = genre.lower() + for key in GENRE_THEMES: + if key in genre_lower: + return GENRE_THEMES[key] + return GENRE_THEMES['default'] + + +def _build_json_ld(episode: dict, novel_config: dict, post_url: str = '') -> str: + """Schema.org Article JSON-LD 생성""" + schema = { + '@context': 'https://schema.org', + '@type': 'Article', + 'headline': f"{novel_config.get('title_ko', '')} {episode.get('episode_num', 0)}화 — {episode.get('title', '')}", + 'description': episode.get('hook', ''), + 'datePublished': datetime.now(timezone.utc).isoformat(), + 'dateModified': datetime.now(timezone.utc).isoformat(), + 'author': { + '@type': 'Person', + 'name': 'The 4th Path' + }, + 'publisher': { + '@type': 'Organization', + 'name': 'The 4th Path', + 'logo': { + '@type': 'ImageObject', + 'url': f'{BLOG_BASE_URL}/logo.png' + } + }, + 'mainEntityOfPage': { + '@type': 'WebPage', + '@id': post_url or BLOG_BASE_URL + }, + 'genre': novel_config.get('genre', ''), + 'isPartOf': { + '@type': 'CreativeWorkSeries', + 'name': novel_config.get('title_ko', ''), + 'position': episode.get('episode_num', 0) + } + } + return ( + '' + ) + + +def _body_to_html(body_text: str) -> str: + """소설 본문 텍스트 → HTML 단락 변환 (빈 줄 기준 분리)""" + paragraphs = [] + for para in body_text.split('\n\n'): + para = para.strip() + if not para: + continue + # 대화문 들여쓰기 처리 + lines = para.split('\n') + html_lines = [] + for line in lines: + line = line.strip() + if not line: + continue + # HTML 특수문자 이스케이프 + line = (line.replace('&', '&') + .replace('<', '<') + .replace('>', '>')) + # 대화문 (따옴표 시작) 스타일 적용 + if line.startswith('"') or line.startswith('"') or line.startswith('"'): + html_lines.append( + f'{line}' + ) + else: + html_lines.append(line) + paragraphs.append('
\n'.join(html_lines)) + + return '\n'.join( + f'

{p}

' + for p in paragraphs if p + ) + + +def convert( + episode: dict, + novel_config: dict, + prev_url: str = '', + next_url: str = '', + save_file: bool = True +) -> str: + """ + 에피소드 + 소설 설정 → Blogger-ready HTML. + data/novels/{novel_id}/episodes/ep{N:03d}_blog.html 저장. + 반환: HTML 문자열 + """ + novel_id = novel_config.get('novel_id', episode.get('novel_id', 'unknown')) + ep_num = episode.get('episode_num', 0) + title = episode.get('title', f'에피소드 {ep_num}') + body_text = episode.get('body', '') + hook = episode.get('hook', '') + genre = novel_config.get('genre', '') + title_ko = novel_config.get('title_ko', '') + + logger.info(f"[{novel_id}] 에피소드 {ep_num} 블로그 변환 시작") + + theme = _get_theme(genre) + bg = theme['bg'] + accent = theme['accent'] + accent_dim = theme['accent_dim'] + card_bg = theme['card_bg'] + text_color = theme['text'] + meta_color = theme['meta'] + nav_bg = theme['nav_bg'] + + # 다음 에피소드 예정일 (publish_schedule 파싱 — 간단 처리) + next_date_str = '다음 회 예고' + try: + schedule = novel_config.get('publish_schedule', '') + # "매주 월/목 09:00" 형식에서 요일 추출 + if schedule: + next_date_str = schedule.replace('매주 ', '').replace('09:00', '').strip() + except Exception: + pass + + # 본문 HTML + body_html = _body_to_html(body_text) + + # JSON-LD + post_url = '' + json_ld = _build_json_ld(episode, novel_config, post_url) + + # 이전/다음 네비게이션 + prev_link = ( + f'← {ep_num - 1}화' + if prev_url and ep_num > 1 + else f'첫 번째 에피소드' + ) + next_link = ( + f'{ep_num + 1}화 →' + if next_url + else f'다음 회 업데이트 예정' + ) + + # 전체 HTML 조립 + html = f"""{json_ld} + + +
+ + +
+ + 연재소설 · 에피소드 {ep_num} + +
+ + +

+ {title_ko} +

+ + +

+ {title} +

+ + +
+ {genre} + · + {novel_config.get('episode_length','')} + · + {datetime.now().strftime('%Y.%m.%d')} +
+ + +
+ {prev_link} + {ep_num}화 + {next_link} +
+ + +
+ {body_html} +
+ + +
+ + +
+

+ 다음 에피소드 예고 · {next_date_str} +

+

+ {hook if hook else '다음 회를 기대해 주세요.'} +

+
+ + + + + +
+ {prev_link} + ↑ 목록 + {next_link} +
+ + +
+

+ {title_ko} 정보 +

+

+ 장르: {genre} · 목표 {novel_config.get('episode_count_target', 20)}화 완결 +

+

+ 연재 일정: {novel_config.get('publish_schedule', '')} · The 4th Path +

+
+ +
""" + + if save_file: + output_dir = BASE_DIR / 'data' / 'novels' / novel_id / 'episodes' + output_dir.mkdir(parents=True, exist_ok=True) + output_path = output_dir / f'ep{ep_num:03d}_blog.html' + try: + output_path.write_text(html, encoding='utf-8') + logger.info(f"블로그 HTML 저장: {output_path}") + except Exception as e: + logger.error(f"블로그 HTML 저장 실패: {e}") + + logger.info(f"[{novel_id}] 에피소드 {ep_num} 블로그 변환 완료") + return html + + +# ─── 직접 실행 테스트 ───────────────────────────────────────────────────────── + +if __name__ == '__main__': + import sys as _sys + logging.basicConfig(level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s') + + # 샘플 테스트 + sample_config = json.loads( + (BASE_DIR / 'config' / 'novels' / 'shadow-protocol.json') + .read_text(encoding='utf-8') + ) + sample_episode = { + 'novel_id': 'shadow-protocol', + 'episode_num': 1, + 'title': '프로토콜', + 'body': '빗소리가 유리창을 두드렸다.\n\n서진은 모니터를 응시했다.', + 'hook': '아리아의 목소리가 처음으로 떨렸다.', + 'key_scenes': [ + '서진이 빗속에서 데이터를 분석하는 장면', + '아리아가 이상 신호를 감지하는 장면', + '감시 드론이 서진의 아파트 창문을 지나가는 장면', + ], + 'summary': '서진은 오라클 시스템에서 숨겨진 프로토콜을 발견한다.', + 'generated_at': '2026-03-26T00:00:00+00:00', + } + html = convert(sample_episode, sample_config) + print(f"HTML 생성 완료: {len(html)}자") diff --git a/bots/novel/novel_manager.py b/bots/novel/novel_manager.py new file mode 100644 index 0000000..698141a --- /dev/null +++ b/bots/novel/novel_manager.py @@ -0,0 +1,507 @@ +""" +novel_manager.py +소설 연재 파이프라인 — 연재 관리 + Telegram 명령어 처리 모듈 +역할: 소설 목록 관리, 에피소드 파이프라인 실행, 스케줄 조정, Telegram 응답 +""" +import json +import logging +import sys +from datetime import datetime, timezone +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = Path(__file__).parent.parent.parent +sys.path.insert(0, str(BASE_DIR / 'bots')) +sys.path.insert(0, str(BASE_DIR / 'bots' / 'novel')) + +logger = logging.getLogger(__name__) +if not logger.handlers: + logs_dir = BASE_DIR / 'logs' + logs_dir.mkdir(exist_ok=True) + handler = logging.FileHandler(logs_dir / 'novel.log', encoding='utf-8') + handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')) + logger.addHandler(handler) + logger.addHandler(logging.StreamHandler()) + logger.setLevel(logging.INFO) + + +# ─── NovelManager ──────────────────────────────────────────────────────────── + +class NovelManager: + """config/novels/*.json 전체를 관리하고 파이프라인을 실행하는 클래스.""" + + def __init__(self): + self.novels_config_dir = BASE_DIR / 'config' / 'novels' + self.novels_data_dir = BASE_DIR / 'data' / 'novels' + self.novels_config_dir.mkdir(parents=True, exist_ok=True) + self.novels_data_dir.mkdir(parents=True, exist_ok=True) + + # ── 소설 목록 조회 ──────────────────────────────────────────────────────── + + def get_all_novels(self) -> list: + """config/novels/*.json 전체 로드""" + novels = [] + for path in sorted(self.novels_config_dir.glob('*.json')): + try: + data = json.loads(path.read_text(encoding='utf-8')) + novels.append(data) + except Exception as e: + logger.error(f"소설 설정 로드 실패 ({path.name}): {e}") + return novels + + def get_active_novels(self) -> list: + """status == 'active' 인 소설만 반환""" + return [n for n in self.get_all_novels() if n.get('status') == 'active'] + + def get_due_novels(self) -> list: + """ + 오늘 발행 예정인 소설 반환. + publish_schedule 예: "매주 월/목 09:00" + """ + today_weekday = datetime.now().weekday() # 0=월, 1=화, ..., 6=일 + _KO_DAY_MAP = { + '월': 0, '화': 1, '수': 2, '목': 3, '금': 4, '토': 5, '일': 6 + } + + due = [] + for novel in self.get_active_novels(): + schedule = novel.get('publish_schedule', '') + try: + # "매주 월/목 09:00" 형식 파싱 + parts = schedule.replace('매주 ', '').split(' ') + days_part = parts[0] if parts else '' + days = [d.strip() for d in days_part.split('/')] + for day in days: + if _KO_DAY_MAP.get(day) == today_weekday: + due.append(novel) + break + except Exception as e: + logger.warning(f"스케줄 파싱 실패 ({novel.get('novel_id')}): {e}") + return due + + # ── 파이프라인 실행 ─────────────────────────────────────────────────────── + + def run_episode_pipeline(self, novel_id: str, + telegram_notify: bool = True) -> bool: + """ + 완전한 에피소드 파이프라인: + 1. NovelWriter.generate_episode() + 2. NovelBlogConverter.convert() + 3. NovelShortsConverter.generate() + 4. publisher_bot으로 블로그 발행 + 5. 성공 시 Telegram 알림 + 반환: 성공 여부 + """ + logger.info(f"[{novel_id}] 에피소드 파이프라인 시작") + + # 소설 설정 로드 + config_path = self.novels_config_dir / f'{novel_id}.json' + if not config_path.exists(): + logger.error(f"소설 설정 없음: {config_path}") + return False + + try: + novel_config = json.loads(config_path.read_text(encoding='utf-8')) + except Exception as e: + logger.error(f"소설 설정 로드 실패: {e}") + return False + + # 데이터 디렉터리 생성 + self.create_novel_dirs(novel_id) + + # ── Step 1: 에피소드 생성 ───────────────────────────────────────────── + episode = None + try: + from novel_writer import NovelWriter + writer = NovelWriter(novel_id) + episode = writer.generate_episode() + if not episode: + logger.error(f"[{novel_id}] 에피소드 생성 실패") + return False + logger.info(f"[{novel_id}] 에피소드 {episode['episode_num']} 생성 완료") + except Exception as e: + logger.error(f"[{novel_id}] Step 1 (에피소드 생성) 실패: {e}") + return False + + # ── Step 2: 블로그 HTML 변환 ────────────────────────────────────────── + html = '' + try: + from novel_blog_converter import convert as blog_convert + html = blog_convert(episode, novel_config, save_file=True) + logger.info(f"[{novel_id}] 블로그 HTML 변환 완료") + except Exception as e: + logger.error(f"[{novel_id}] Step 2 (블로그 변환) 실패: {e}") + + # ── Step 3: 쇼츠 영상 생성 ─────────────────────────────────────────── + shorts_path = '' + try: + from novel_shorts_converter import NovelShortsConverter + converter = NovelShortsConverter() + shorts_path = converter.generate(episode, novel_config) + if shorts_path: + logger.info(f"[{novel_id}] 쇼츠 생성 완료: {shorts_path}") + else: + logger.warning(f"[{novel_id}] 쇼츠 생성 실패 (계속 진행)") + except Exception as e: + logger.error(f"[{novel_id}] Step 3 (쇼츠 생성) 실패: {e}") + + # ── Step 4: 블로그 발행 ─────────────────────────────────────────────── + publish_ok = False + if html: + try: + publish_ok = self._publish_episode(episode, novel_config, html) + if publish_ok: + logger.info(f"[{novel_id}] 블로그 발행 완료") + else: + logger.warning(f"[{novel_id}] 블로그 발행 실패") + except Exception as e: + logger.error(f"[{novel_id}] Step 4 (발행) 실패: {e}") + + # ── Step 5: Telegram 알림 ───────────────────────────────────────────── + if telegram_notify: + try: + ep_num = episode.get('episode_num', 0) + title = episode.get('title', '') + msg = ( + f"소설 연재 완료!\n" + f"제목: {novel_config.get('title_ko', novel_id)}\n" + f"에피소드: {ep_num}화 — {title}\n" + f"블로그: {'발행 완료' if publish_ok else '발행 실패'}\n" + f"쇼츠: {'생성 완료' if shorts_path else '생성 실패'}" + ) + self._send_telegram(msg) + except Exception as e: + logger.warning(f"Telegram 알림 실패: {e}") + + success = episode is not None + logger.info(f"[{novel_id}] 파이프라인 완료 (성공={success})") + return success + + # ── 소설 상태 조회 ──────────────────────────────────────────────────────── + + def get_novel_status(self, novel_id: str) -> dict: + """소설 현황 반환 (에피소드 수, 마지막 발행일, 다음 예정일 등)""" + config_path = self.novels_config_dir / f'{novel_id}.json' + if not config_path.exists(): + return {} + + try: + config = json.loads(config_path.read_text(encoding='utf-8')) + except Exception as e: + logger.error(f"소설 설정 로드 실패: {e}") + return {} + + episodes_dir = self.novels_data_dir / novel_id / 'episodes' + ep_files = list(episodes_dir.glob('ep*.json')) if episodes_dir.exists() else [] + # 요약/블로그 파일 제외 (ep001.json 만 카운트) + ep_files = [ + f for f in ep_files + if '_summary' not in f.name and '_blog' not in f.name + ] + + last_ep_date = '' + if config.get('episode_log'): + last_log = config['episode_log'][-1] + last_ep_date = last_log.get('generated_at', '')[:10] + + return { + 'novel_id': novel_id, + 'title_ko': config.get('title_ko', ''), + 'status': config.get('status', 'unknown'), + 'current_episode': config.get('current_episode', 0), + 'episode_count_target': config.get('episode_count_target', 0), + 'episode_files': len(ep_files), + 'last_published': last_ep_date, + 'publish_schedule': config.get('publish_schedule', ''), + 'genre': config.get('genre', ''), + } + + def list_novels_text(self) -> str: + """Telegram용 소설 목록 텍스트 반환""" + novels = self.get_all_novels() + if not novels: + return '등록된 소설이 없습니다.' + + lines = ['소설 목록:\n'] + for n in novels: + status_label = '연재중' if n.get('status') == 'active' else '중단' + lines.append( + f"[{status_label}] {n.get('title_ko', n.get('novel_id', ''))}\n" + f" 장르: {n.get('genre', '')} | " + f"{n.get('current_episode', 0)}/{n.get('episode_count_target', 0)}화 | " + f"{n.get('publish_schedule', '')}\n" + ) + return '\n'.join(lines) + + # ── 디렉터리 생성 ───────────────────────────────────────────────────────── + + def create_novel_dirs(self, novel_id: str): + """data/novels/{novel_id}/episodes/, shorts/, images/ 폴더 생성""" + base = self.novels_data_dir / novel_id + for sub in ['episodes', 'shorts', 'images']: + (base / sub).mkdir(parents=True, exist_ok=True) + logger.info(f"[{novel_id}] 데이터 디렉터리 생성: {base}") + + # ── 내부 헬퍼 ───────────────────────────────────────────────────────────── + + def _publish_episode(self, episode: dict, novel_config: dict, + html: str) -> bool: + """publisher_bot을 통해 블로그 발행""" + try: + import publisher_bot + novel_id = novel_config.get('novel_id', '') + ep_num = episode.get('episode_num', 0) + title_ko = novel_config.get('title_ko', '') + + article = { + 'title': f"{title_ko} {ep_num}화 — {episode.get('title', '')}", + 'body': html, + '_body_is_html': True, + '_html_content': html, + 'corner': '연재소설', + 'slug': f"{novel_id}-ep{ep_num:03d}", + 'labels': ['연재소설', title_ko, f'에피소드{ep_num}'], + } + return publisher_bot.publish(article) + except ImportError: + logger.warning("publisher_bot 없음 — 발행 건너뜀") + return False + except Exception as e: + logger.error(f"발행 실패: {e}") + return False + + def _send_telegram(self, message: str): + """Telegram 메시지 전송""" + try: + import telegram_bot + telegram_bot.send_message(message) + except ImportError: + logger.warning("telegram_bot 없음 — 알림 건너뜀") + except Exception as e: + logger.warning(f"Telegram 전송 실패: {e}") + + def _update_novel_status(self, novel_id: str, status: str) -> bool: + """소설 status 필드 업데이트 (active / paused)""" + config_path = self.novels_config_dir / f'{novel_id}.json' + if not config_path.exists(): + return False + try: + config = json.loads(config_path.read_text(encoding='utf-8')) + config['status'] = status + config_path.write_text( + json.dumps(config, ensure_ascii=False, indent=2), + encoding='utf-8' + ) + logger.info(f"[{novel_id}] status 변경: {status}") + return True + except Exception as e: + logger.error(f"소설 status 업데이트 실패: {e}") + return False + + def _find_novel_by_title(self, title_query: str) -> str: + """소설 제목(한국어 or 영어) 또는 novel_id로 검색 — novel_id 반환""" + title_query = title_query.strip() + for novel in self.get_all_novels(): + if (title_query in novel.get('title_ko', '') + or title_query in novel.get('title', '') + or title_query == novel.get('novel_id', '')): + return novel.get('novel_id', '') + return '' + + def run_all(self) -> list: + """오늘 발행 예정인 모든 활성 소설 파이프라인 실행 (스케줄러용)""" + results = [] + for novel in self.get_due_novels(): + novel_id = novel.get('novel_id', '') + ok = self.run_episode_pipeline(novel_id, telegram_notify=True) + results.append({'novel_id': novel_id, 'success': ok}) + return results + + +# ─── Telegram 명령 처리 함수 (scheduler.py에서 호출) ───────────────────────── + +def handle_novel_command(text: str) -> str: + """ + Telegram 소설 명령어 처리. + + 지원 명령: + "소설 새로 만들기" + "소설 목록" + "소설 {제목} 다음 에피소드" + "소설 {제목} 현황" + "소설 {제목} 중단" + "소설 {제목} 재개" + + 반환: 응답 문자열 + """ + manager = NovelManager() + text = text.strip() + + # ── "소설 목록" ─────────────────────────────────────────────────────────── + if text in ('소설 목록', '소설목록'): + return manager.list_novels_text() + + # ── "소설 새로 만들기" ──────────────────────────────────────────────────── + if '새로 만들기' in text or '새로만들기' in text: + return ( + '새 소설 설정 방법:\n\n' + '1. config/novels/ 폴더에 {novel_id}.json 파일 생성\n' + '2. 필수 필드: novel_id, title, title_ko, genre,\n' + ' setting, characters, base_story, publish_schedule\n' + '3. status: "active" 로 설정하면 자동 연재 시작\n\n' + '예시: config/novels/shadow-protocol.json 참고' + ) + + # ── "소설 {제목} 다음 에피소드" ─────────────────────────────────────────── + if '다음 에피소드' in text: + title_query = (text.replace('소설', '', 1) + .replace('다음 에피소드', '') + .strip()) + if not title_query: + return '소설 제목을 입력해 주세요.\n예: 소설 그림자 프로토콜 다음 에피소드' + + novel_id = manager._find_novel_by_title(title_query) + if not novel_id: + return ( + f'"{title_query}" 소설을 찾을 수 없습니다.\n' + '소설 목록을 확인해 주세요.' + ) + + status_info = manager.get_novel_status(novel_id) + if status_info.get('status') != 'active': + return f'"{title_query}" 소설은 현재 연재 중단 상태입니다.' + + try: + ok = manager.run_episode_pipeline(novel_id, telegram_notify=False) + if ok: + updated = manager.get_novel_status(novel_id) + ep = updated.get('current_episode', 0) + return ( + f"에피소드 {ep}화 생성 및 발행 완료!\n" + f"소설: {updated.get('title_ko', novel_id)}" + ) + else: + return '에피소드 생성 실패. 로그를 확인해 주세요.' + except Exception as e: + logger.error(f"에피소드 파이프라인 오류: {e}") + return f'오류 발생: {e}' + + # ── "소설 {제목} 현황" ──────────────────────────────────────────────────── + if '현황' in text: + title_query = text.replace('소설', '', 1).replace('현황', '').strip() + if not title_query: + return '소설 제목을 입력해 주세요.\n예: 소설 그림자 프로토콜 현황' + + novel_id = manager._find_novel_by_title(title_query) + if not novel_id: + return f'"{title_query}" 소설을 찾을 수 없습니다.' + + s = manager.get_novel_status(novel_id) + if not s: + return f'"{title_query}" 현황을 불러올 수 없습니다.' + + return ( + f"소설 현황: {s.get('title_ko', novel_id)}\n\n" + f"상태: {s.get('status', '')}\n" + f"현재 에피소드: {s.get('current_episode', 0)}화 / " + f"목표 {s.get('episode_count_target', 0)}화\n" + f"마지막 발행: {s.get('last_published', '없음')}\n" + f"연재 일정: {s.get('publish_schedule', '')}\n" + f"장르: {s.get('genre', '')}" + ) + + # ── "소설 {제목} 중단" ──────────────────────────────────────────────────── + if '중단' in text: + title_query = text.replace('소설', '', 1).replace('중단', '').strip() + if not title_query: + return '소설 제목을 입력해 주세요.\n예: 소설 그림자 프로토콜 중단' + + novel_id = manager._find_novel_by_title(title_query) + if not novel_id: + return f'"{title_query}" 소설을 찾을 수 없습니다.' + + ok = manager._update_novel_status(novel_id, 'paused') + if ok: + try: + config = json.loads( + (manager.novels_config_dir / f'{novel_id}.json') + .read_text(encoding='utf-8') + ) + return f'"{config.get("title_ko", novel_id)}" 연재를 일시 중단했습니다.' + except Exception: + return '중단 처리 완료.' + else: + return '중단 처리 실패. 로그를 확인해 주세요.' + + # ── "소설 {제목} 재개" ──────────────────────────────────────────────────── + if '재개' in text: + title_query = text.replace('소설', '', 1).replace('재개', '').strip() + if not title_query: + return '소설 제목을 입력해 주세요.\n예: 소설 그림자 프로토콜 재개' + + novel_id = manager._find_novel_by_title(title_query) + if not novel_id: + return f'"{title_query}" 소설을 찾을 수 없습니다.' + + ok = manager._update_novel_status(novel_id, 'active') + if ok: + try: + config = json.loads( + (manager.novels_config_dir / f'{novel_id}.json') + .read_text(encoding='utf-8') + ) + return ( + f'"{config.get("title_ko", novel_id)}" 연재를 재개합니다.\n' + f'연재 일정: {config.get("publish_schedule", "")}' + ) + except Exception: + return '재개 처리 완료.' + else: + return '재개 처리 실패. 로그를 확인해 주세요.' + + # ── 알 수 없는 명령 ─────────────────────────────────────────────────────── + return ( + '소설 명령어 목록:\n\n' + '소설 목록\n' + '소설 새로 만들기\n' + '소설 {제목} 다음 에피소드\n' + '소설 {제목} 현황\n' + '소설 {제목} 중단\n' + '소설 {제목} 재개' + ) + + +# ─── 직접 실행 테스트 ───────────────────────────────────────────────────────── + +if __name__ == '__main__': + logging.basicConfig(level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s') + + manager = NovelManager() + + print("=== 전체 소설 목록 ===") + print(manager.list_novels_text()) + + print("\n=== 활성 소설 ===") + for n in manager.get_active_novels(): + print(f" - {n.get('title_ko')} ({n.get('novel_id')})") + + print("\n=== 오늘 발행 예정 소설 ===") + due = manager.get_due_novels() + if due: + for n in due: + print(f" - {n.get('title_ko')}") + else: + print(" (없음)") + + print("\n=== shadow-protocol 현황 ===") + status = manager.get_novel_status('shadow-protocol') + print(json.dumps(status, ensure_ascii=False, indent=2)) + + print("\n=== Telegram 명령 테스트 ===") + for cmd in ['소설 목록', '소설 그림자 프로토콜 현황', '소설 잘못된제목 현황']: + print(f"\n명령: {cmd}") + print(handle_novel_command(cmd)) diff --git a/bots/novel/novel_shorts_converter.py b/bots/novel/novel_shorts_converter.py new file mode 100644 index 0000000..5a95a22 --- /dev/null +++ b/bots/novel/novel_shorts_converter.py @@ -0,0 +1,684 @@ +""" +novel_shorts_converter.py +소설 연재 파이프라인 — 에피소드 KEY_SCENES → 쇼츠 영상 생성 모듈 +역할: key_scenes 3개를 기반으로 engine.json 설정에 따라 영상 생성 +출력: data/novels/{novel_id}/shorts/ep{N:03d}_shorts.mp4 + +지원 모드: + ffmpeg_slides — DALL-E 이미지 + TTS + Pillow 슬라이드 (기본, 비용 0원) + seedance — Seedance 2.0 API로 시네마틱 영상 생성 (유료) +""" +import json +import logging +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = Path(__file__).parent.parent.parent +sys.path.insert(0, str(BASE_DIR / 'bots')) +sys.path.insert(0, str(BASE_DIR / 'bots' / 'converters')) + +logger = logging.getLogger(__name__) + +FFMPEG = os.getenv('FFMPEG_PATH', 'ffmpeg') +OPENAI_API_KEY = os.getenv('OPENAI_API_KEY', '') +SEEDANCE_API_KEY = os.getenv('SEEDANCE_API_KEY', '') + +# ─── EngineLoader / fallback ───────────────────────────────────────────────── + +try: + from engine_loader import EngineLoader as _EngineLoader + _engine_loader_available = True +except ImportError: + _engine_loader_available = False + + +def _make_fallback_writer(): + """engine_loader 없을 때 anthropic SDK 직접 사용""" + import anthropic + client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY', '')) + + class _DirectWriter: + def write(self, prompt: str, system: str = '') -> str: + msg = client.messages.create( + model='claude-opus-4-6', + max_tokens=500, + system=system or '당신은 한국어 연재소설 작가입니다.', + messages=[{'role': 'user', 'content': prompt}], + ) + return msg.content[0].text + + return _DirectWriter() + + +# shorts_converter 유틸 임포트 (재사용) +try: + from shorts_converter import ( + make_clip, + concat_clips_xfade, + mix_bgm, + burn_subtitles, + synthesize_section, + solid_background, + _load_font, + _text_size, + _draw_gradient_overlay, + ) + _shorts_utils_available = True +except ImportError: + _shorts_utils_available = False + logger.warning("shorts_converter 임포트 실패 — ffmpeg_slides 모드 제한됨") + + +# ─── 이미지 생성 헬퍼 ───────────────────────────────────────────────────────── + +def _generate_dalle_image(prompt: str, save_path: str) -> bool: + """DALL-E 3로 장면 이미지 생성 (1024×1792)""" + if not OPENAI_API_KEY: + logger.warning("OPENAI_API_KEY 없음 — 단색 배경 사용") + return False + try: + import io + import requests as req + from openai import OpenAI + from PIL import Image + + client = OpenAI(api_key=OPENAI_API_KEY) + full_prompt = prompt + ' No text, no letters, no numbers, no watermarks. Vertical 9:16 cinematic.' + response = client.images.generate( + model='dall-e-3', + prompt=full_prompt, + size='1024x1792', + quality='standard', + n=1, + ) + img_url = response.data[0].url + img_bytes = req.get(img_url, timeout=30).content + img = Image.open(io.BytesIO(img_bytes)).convert('RGB') + img = img.resize((1080, 1920)) + img.save(save_path) + logger.info(f"DALL-E 이미지 생성: {save_path}") + return True + except Exception as e: + logger.warning(f"DALL-E 이미지 생성 실패: {e}") + return False + + +def _make_solid_slide(save_path: str, color=(10, 10, 13)): + """단색 슬라이드 PNG 생성""" + try: + from PIL import Image + img = Image.new('RGB', (1080, 1920), color) + img.save(save_path) + except Exception as e: + logger.error(f"단색 슬라이드 생성 실패: {e}") + + +def _make_text_slide(save_path: str, text: str, bg_color=(10, 10, 13), + accent_color=(200, 168, 78)): + """텍스트 오버레이 슬라이드 PNG 생성""" + try: + from PIL import Image, ImageDraw + img = Image.new('RGB', (1080, 1920), bg_color) + draw = ImageDraw.Draw(img) + W, H = 1080, 1920 + + # 상단 강조선 + draw.rectangle([0, 0, W, 6], fill=accent_color) + + # 텍스트 + font = _load_font(52, bold=True) if _shorts_utils_available else None + if font: + words = text.split() + lines = [] + current = '' + for word in words: + test = (current + ' ' + word).strip() + w, _ = _text_size(draw, test, font) + if w <= W - 120: + current = test + else: + if current: + lines.append(current) + current = word + if current: + lines.append(current) + + y = H // 2 - len(lines) * 60 // 2 + for line in lines[:5]: + lw, lh = _text_size(draw, line, font) + draw.text(((W - lw) // 2, y), line, font=font, fill=(255, 255, 255)) + y += lh + 16 + + # 하단 강조선 + draw.rectangle([0, H - 6, W, H], fill=accent_color) + img.save(save_path) + except Exception as e: + logger.error(f"텍스트 슬라이드 생성 실패: {e}") + + +# ─── Seedance API 헬퍼 ──────────────────────────────────────────────────────── + +def _call_seedance_api(prompt: str, duration: str = '10s', + resolution: str = '1080x1920') -> str: + """ + Seedance 2.0 API 호출 → 영상 다운로드 후 임시 경로 반환. + 실패 시 '' 반환. + """ + if not SEEDANCE_API_KEY: + logger.warning("SEEDANCE_API_KEY 없음 — seedance 모드 불가") + return '' + try: + import requests as req + # engine.json에서 api_url 읽기 + engine_config_path = BASE_DIR / 'config' / 'engine.json' + api_url = 'https://api.seedance2.ai/v1/generate' + if engine_config_path.exists(): + cfg = json.loads(engine_config_path.read_text(encoding='utf-8')) + api_url = (cfg.get('video_generation', {}) + .get('options', {}) + .get('seedance', {}) + .get('api_url', api_url)) + + headers = { + 'Authorization': f'Bearer {SEEDANCE_API_KEY}', + 'Content-Type': 'application/json', + } + payload = { + 'prompt': prompt, + 'resolution': resolution, + 'duration': duration, + 'audio': True, + } + logger.info(f"Seedance API 요청: {prompt[:80]}...") + resp = req.post(api_url, json=payload, headers=headers, timeout=120) + resp.raise_for_status() + data = resp.json() + + # 영상 URL 파싱 (실제 API 응답 구조에 따라 조정 필요) + video_url = data.get('url') or data.get('video_url') or data.get('output') + if not video_url: + logger.error(f"Seedance 응답에 URL 없음: {data}") + return '' + + # 영상 다운로드 + tmp_path = tempfile.mktemp(suffix='.mp4') + video_resp = req.get(video_url, timeout=180) + video_resp.raise_for_status() + Path(tmp_path).write_bytes(video_resp.content) + logger.info(f"Seedance 영상 다운로드 완료: {tmp_path}") + return tmp_path + + except Exception as e: + logger.error(f"Seedance API 호출 실패: {e}") + return '' + + +def _run_ffmpeg(args: list) -> bool: + """ffmpeg 실행 헬퍼""" + cmd = [FFMPEG, '-y', '-loglevel', 'error'] + args + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) + if result.returncode != 0: + logger.error(f"ffmpeg 오류: {result.stderr[-400:]}") + return result.returncode == 0 + except Exception as e: + logger.error(f"ffmpeg 실행 실패: {e}") + return False + + +def _get_clip_duration(mp4_path: str) -> float: + """ffprobe로 영상 길이(초) 측정""" + try: + import subprocess + result = subprocess.run( + ['ffprobe', '-v', 'quiet', '-print_format', 'json', + '-show_format', mp4_path], + capture_output=True, text=True, timeout=10 + ) + data = json.loads(result.stdout) + return float(data['format']['duration']) + except Exception: + return 10.0 + + +# ─── 테마 색상 헬퍼 ─────────────────────────────────────────────────────────── + +_GENRE_COLORS = { + 'sci-fi': {'bg': (10, 15, 30), 'accent': (0, 188, 212)}, + 'thriller': {'bg': (10, 10, 13), 'accent': (191, 58, 58)}, + 'fantasy': {'bg': (15, 10, 30), 'accent': (200, 168, 78)}, + 'romance': {'bg': (255, 245, 240), 'accent': (216, 90, 48)}, + 'default': {'bg': (10, 10, 13), 'accent': (200, 168, 78)}, +} + +def _genre_colors(genre: str) -> dict: + genre_lower = genre.lower() + for key in _GENRE_COLORS: + if key in genre_lower: + return _GENRE_COLORS[key] + return _GENRE_COLORS['default'] + + +# ─── NovelShortsConverter ──────────────────────────────────────────────────── + +class NovelShortsConverter: + """소설 에피소드 key_scenes → 쇼츠 MP4 생성""" + + def __init__(self, engine=None): + """engine: EngineLoader 인스턴스 (없으면 내부에서 결정)""" + if engine is not None: + self.engine = engine + elif _engine_loader_available: + self.engine = _EngineLoader() + else: + self.engine = None + + # video_generation provider 결정 + self.video_provider = self._get_video_provider() + + # 번역용 writer (씬 → 영어 프롬프트) + if self.engine is not None: + try: + self.writer = self.engine.get_writer() + except Exception: + self.writer = _make_fallback_writer() + else: + self.writer = _make_fallback_writer() + + def _get_video_provider(self) -> str: + """engine.json에서 video_generation.provider 읽기""" + try: + cfg_path = BASE_DIR / 'config' / 'engine.json' + if cfg_path.exists(): + cfg = json.loads(cfg_path.read_text(encoding='utf-8')) + return cfg.get('video_generation', {}).get('provider', 'ffmpeg_slides') + except Exception: + pass + return 'ffmpeg_slides' + + # ── 공개 API ────────────────────────────────────────────────────────────── + + def generate(self, episode: dict, novel_config: dict) -> str: + """ + episode의 key_scenes 3개로 쇼츠 생성. + 반환: MP4 경로 (data/novels/{novel_id}/shorts/ep{N:03d}_shorts.mp4) + 실패 시 '' 반환. + """ + novel_id = novel_config.get('novel_id', episode.get('novel_id', 'unknown')) + ep_num = episode.get('episode_num', 0) + key_scenes = episode.get('key_scenes', []) + + if not key_scenes: + logger.error(f"[{novel_id}] key_scenes 없음 — 쇼츠 생성 불가") + return '' + + # 출력 디렉터리 준비 + shorts_dir = BASE_DIR / 'data' / 'novels' / novel_id / 'shorts' + shorts_dir.mkdir(parents=True, exist_ok=True) + output_path = str(shorts_dir / f'ep{ep_num:03d}_shorts.mp4') + + logger.info(f"[{novel_id}] 에피소드 {ep_num} 쇼츠 생성 시작 (모드: {self.video_provider})") + + try: + if self.video_provider == 'seedance' and SEEDANCE_API_KEY: + result = self._generate_seedance(episode, novel_config, output_path) + else: + if self.video_provider == 'seedance': + logger.warning("SEEDANCE_API_KEY 없음 — ffmpeg_slides 모드로 대체") + result = self._generate_ffmpeg_slides(episode, novel_config, output_path) + + if result: + logger.info(f"[{novel_id}] 쇼츠 생성 완료: {output_path}") + else: + logger.error(f"[{novel_id}] 쇼츠 생성 실패") + + return output_path if result else '' + + except Exception as e: + logger.error(f"[{novel_id}] 쇼츠 생성 중 예외: {e}") + return '' + + # ── ffmpeg_slides 모드 ──────────────────────────────────────────────────── + + def _generate_ffmpeg_slides(self, episode: dict, novel_config: dict, + output_path: str) -> bool: + """ + ffmpeg_slides 모드: + 인트로 슬라이드 + key_scene 1~3 (DALL-E 이미지 + TTS) + 아웃트로 + concat + """ + if not _shorts_utils_available: + logger.error("shorts_converter 유틸 없음 — ffmpeg_slides 불가") + return False + + novel_id = novel_config.get('novel_id', '') + ep_num = episode.get('episode_num', 0) + key_scenes = episode.get('key_scenes', []) + hook = episode.get('hook', '') + title_ko = novel_config.get('title_ko', '') + genre = novel_config.get('genre', '') + colors = _genre_colors(genre) + bg_color = colors['bg'] + accent_color = colors['accent'] + + # 임시 작업 디렉터리 + tmp_dir = Path(tempfile.mkdtemp(prefix=f'novel_{novel_id}_ep{ep_num}_')) + + try: + clips = [] + + # ── 인트로 슬라이드 (소설 제목 + 에피소드 번호) ────────────────── + intro_slide = str(tmp_dir / 'intro.png') + intro_text = f"{title_ko}\n{ep_num}화" + _make_text_slide(intro_slide, f"{title_ko} · {ep_num}화", + bg_color, accent_color) + intro_audio = str(tmp_dir / 'intro.wav') + synthesize_section(f"{title_ko} {ep_num}화", intro_audio, + 'ko-KR-Wavenet-A', 1.0) + intro_mp4 = str(tmp_dir / 'intro.mp4') + dur = make_clip(intro_slide, intro_audio, intro_mp4) + if dur > 0: + clips.append({'mp4': intro_mp4, 'duration': dur}) + + # ── key_scene 슬라이드 1~3 ──────────────────────────────────────── + images_dir = BASE_DIR / 'data' / 'novels' / novel_id / 'images' + images_dir.mkdir(parents=True, exist_ok=True) + + for i, scene in enumerate(key_scenes[:3], 1): + if not scene: + continue + + # DALL-E 이미지 (또는 텍스트 슬라이드 폴백) + scene_slide = str(tmp_dir / f'scene{i}.png') + img_prompt = self._scene_to_image_prompt(scene, novel_config) + img_generated = _generate_dalle_image(img_prompt, scene_slide) + if not img_generated: + _make_text_slide(scene_slide, scene, bg_color, accent_color) + + # 생성된 이미지를 data/images에도 저장 (캐릭터 일관성 참조용) + perm_img = str(images_dir / f'ep{ep_num:03d}_scene{i}.png') + if img_generated and Path(scene_slide).exists(): + import shutil + shutil.copy2(scene_slide, perm_img) + + # TTS + scene_audio = str(tmp_dir / f'scene{i}.wav') + synthesize_section(scene, scene_audio, 'ko-KR-Wavenet-A', 1.0) + + # 클립 생성 + scene_mp4 = str(tmp_dir / f'scene{i}.mp4') + dur = make_clip(scene_slide, scene_audio, scene_mp4) + if dur > 0: + clips.append({'mp4': scene_mp4, 'duration': dur}) + + # ── 아웃트로 슬라이드 (훅 문장 + 다음 에피소드 예고) ────────────── + outro_slide = str(tmp_dir / 'outro.png') + outro_text = hook if hook else '다음 회를 기대해 주세요' + _make_text_slide(outro_slide, outro_text, bg_color, accent_color) + outro_audio = str(tmp_dir / 'outro.wav') + synthesize_section(outro_text, outro_audio, 'ko-KR-Wavenet-A', 1.0) + outro_mp4 = str(tmp_dir / 'outro.mp4') + dur = make_clip(outro_slide, outro_audio, outro_mp4) + if dur > 0: + clips.append({'mp4': outro_mp4, 'duration': dur}) + + if not clips: + logger.error("생성된 클립 없음") + return False + + # ── 클립 결합 ──────────────────────────────────────────────────── + concat_mp4 = str(tmp_dir / 'concat.mp4') + ok = concat_clips_xfade(clips, concat_mp4, transition='fade', trans_dur=0.5) + if not ok: + logger.error("클립 결합 실패") + return False + + # ── BGM 믹스 ───────────────────────────────────────────────────── + bgm_path = str(BASE_DIR / 'assets' / 'bgm.mp3') + bgm_mp4 = str(tmp_dir / 'bgm_mixed.mp4') + mix_bgm(concat_mp4, bgm_path, bgm_mp4, volume=0.08) + final_source = bgm_mp4 if Path(bgm_mp4).exists() else concat_mp4 + + # ── 최종 출력 복사 ──────────────────────────────────────────────── + import shutil + shutil.copy2(final_source, output_path) + return True + + except Exception as e: + logger.error(f"ffmpeg_slides 생성 중 예외: {e}") + return False + finally: + # 임시 파일 정리 + try: + import shutil + shutil.rmtree(tmp_dir, ignore_errors=True) + except Exception: + pass + + # ── seedance 모드 ───────────────────────────────────────────────────────── + + def _generate_seedance(self, episode: dict, novel_config: dict, + output_path: str) -> bool: + """ + seedance 모드: + key_scene → 영어 Seedance 프롬프트 변환 → API 호출 → 클립 concat + 인트로2초 + 씬10초×3 + 아웃트로3초 = 35초 구성 + """ + novel_id = novel_config.get('novel_id', '') + ep_num = episode.get('episode_num', 0) + key_scenes = episode.get('key_scenes', []) + hook = episode.get('hook', '') + title_ko = novel_config.get('title_ko', '') + genre = novel_config.get('genre', '') + colors = _genre_colors(genre) + bg_color = colors['bg'] + accent_color = colors['accent'] + + tmp_dir = Path(tempfile.mkdtemp(prefix=f'novel_seedance_{novel_id}_ep{ep_num}_')) + + try: + clip_paths = [] + + # ── 인트로 슬라이드 (2초, 정적 이미지) ────────────────────────── + intro_slide = str(tmp_dir / 'intro.png') + _make_text_slide(intro_slide, f"{title_ko} · {ep_num}화", + bg_color, accent_color) + intro_mp4 = str(tmp_dir / 'intro.mp4') + if _run_ffmpeg([ + '-loop', '1', '-i', intro_slide, + '-c:v', 'libx264', '-t', '2', + '-pix_fmt', 'yuv420p', + '-vf', 'scale=1080:1920', + intro_mp4, + ]): + clip_paths.append({'mp4': intro_mp4, 'duration': 2.0}) + + # ── key_scene 클립 (Seedance 10초×3) ───────────────────────────── + for i, scene in enumerate(key_scenes[:3], 1): + if not scene: + continue + en_prompt = self._scene_to_seedance_prompt(scene, novel_config) + seedance_mp4 = _call_seedance_api(en_prompt, duration='10s') + + if seedance_mp4 and Path(seedance_mp4).exists(): + # 영구 저장 + images_dir = BASE_DIR / 'data' / 'novels' / novel_id / 'images' + images_dir.mkdir(parents=True, exist_ok=True) + perm_clip = str(images_dir / f'ep{ep_num:03d}_scene{i}_seedance.mp4') + import shutil + shutil.copy2(seedance_mp4, perm_clip) + clip_paths.append({'mp4': seedance_mp4, 'duration': 10.0}) + else: + # Seedance 실패 시 DALL-E 슬라이드 폴백 + logger.warning(f"장면 {i} Seedance 실패 — 이미지 슬라이드 대체") + fallback_slide = str(tmp_dir / f'scene{i}_fallback.png') + img_prompt = self._scene_to_image_prompt(scene, novel_config) + img_ok = _generate_dalle_image(img_prompt, fallback_slide) + if not img_ok: + _make_text_slide(fallback_slide, scene, bg_color, accent_color) + fallback_mp4 = str(tmp_dir / f'scene{i}_fallback.mp4') + if _run_ffmpeg([ + '-loop', '1', '-i', fallback_slide, + '-c:v', 'libx264', '-t', '10', + '-pix_fmt', 'yuv420p', + '-vf', 'scale=1080:1920', + fallback_mp4, + ]): + clip_paths.append({'mp4': fallback_mp4, 'duration': 10.0}) + + # ── 아웃트로 (3초, 정적 이미지) ────────────────────────────────── + outro_slide = str(tmp_dir / 'outro.png') + outro_text = hook if hook else '다음 회를 기대해 주세요' + _make_text_slide(outro_slide, outro_text, bg_color, accent_color) + outro_mp4 = str(tmp_dir / 'outro.mp4') + if _run_ffmpeg([ + '-loop', '1', '-i', outro_slide, + '-c:v', 'libx264', '-t', '3', + '-pix_fmt', 'yuv420p', + '-vf', 'scale=1080:1920', + outro_mp4, + ]): + clip_paths.append({'mp4': outro_mp4, 'duration': 3.0}) + + if not clip_paths: + logger.error("Seedance 모드: 생성된 클립 없음") + return False + + # ── concat (xfade) ──────────────────────────────────────────────── + if _shorts_utils_available: + concat_mp4 = str(tmp_dir / 'concat.mp4') + ok = concat_clips_xfade(clip_paths, concat_mp4, + transition='fade', trans_dur=0.5) + else: + # 단순 concat fallback + concat_mp4 = str(tmp_dir / 'concat.mp4') + list_file = str(tmp_dir / 'clips.txt') + with open(list_file, 'w') as f: + for c in clip_paths: + f.write(f"file '{c['mp4']}'\n") + ok = _run_ffmpeg([ + '-f', 'concat', '-safe', '0', '-i', list_file, + '-c', 'copy', concat_mp4, + ]) + + if not ok: + logger.error("클립 결합 실패") + return False + + import shutil + shutil.copy2(concat_mp4, output_path) + return True + + except Exception as e: + logger.error(f"Seedance 모드 생성 중 예외: {e}") + return False + finally: + try: + import shutil + shutil.rmtree(tmp_dir, ignore_errors=True) + except Exception: + pass + + # ── 프롬프트 변환 메서드 ────────────────────────────────────────────────── + + def _scene_to_seedance_prompt(self, scene: str, novel_config: dict) -> str: + """한국어 장면 묘사 → 영어 Seedance 프롬프트 변환""" + genre = novel_config.get('genre', 'thriller') + atmosphere = novel_config.get('setting', {}).get('atmosphere', '') + + prompt = f"""다음 한국어 소설 장면 묘사를 Seedance 2.0 AI 영상 생성용 영어 프롬프트로 변환하세요. + +장르: {genre} +분위기: {atmosphere} +장면: {scene} + +변환 규칙: +- 영어로만 작성 +- 시각적이고 구체적인 묘사 (인물, 배경, 조명, 날씨, 카메라 움직임) +- 분위기 키워드 포함 (cinematic, neo-noir 등 장르에 맞게) +- "9:16 vertical" 포함 +- No text overlays, no watermarks +- 3~5문장, 100단어 이내 + +영어 프롬프트만 출력 (설명 없이):""" + + try: + result = self.writer.write(prompt, system='You are a cinematic AI video prompt engineer.') + return result.strip() + except Exception as e: + logger.error(f"Seedance 프롬프트 변환 실패: {e}") + # 폴백: 간단 번역 구조 + genre_key = 'neo-noir sci-fi' if 'sci-fi' in genre else genre + return ( + f"Cinematic scene: {scene[:80]}. " + f"{genre_key} atmosphere. " + f"Dramatic lighting, rain-soaked streets of Seoul 2040. " + f"Vertical 9:16 cinematic shot. No text, no watermarks." + ) + + def _scene_to_image_prompt(self, scene: str, novel_config: dict) -> str: + """DALL-E용 이미지 프롬프트 생성""" + genre = novel_config.get('genre', 'thriller') + world = novel_config.get('setting', {}).get('world', '') + atmosphere = novel_config.get('setting', {}).get('atmosphere', '') + + prompt = f"""소설 장면 묘사를 DALL-E 3 이미지 생성용 영어 프롬프트로 변환하세요. + +세계관: {world} +분위기: {atmosphere} +장면: {scene} + +규칙: +- 영어, 2~3문장 +- 세로형(1024×1792) 구도 +- 장르({genre})에 맞는 분위기 +- No text, no letters, no watermarks + +영어 프롬프트만 출력:""" + + try: + result = self.writer.write(prompt, system='You are a visual prompt engineer for DALL-E.') + return result.strip() + except Exception as e: + logger.error(f"이미지 프롬프트 변환 실패: {e}") + return ( + f"Cinematic vertical illustration: {scene[:80]}. " + f"Dark atmospheric lighting. " + f"No text, no watermarks. Vertical 9:16." + ) + + +# ─── 직접 실행 테스트 ───────────────────────────────────────────────────────── + +if __name__ == '__main__': + logging.basicConfig(level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s') + + sample_config = json.loads( + (BASE_DIR / 'config' / 'novels' / 'shadow-protocol.json') + .read_text(encoding='utf-8') + ) + sample_episode = { + 'novel_id': 'shadow-protocol', + 'episode_num': 1, + 'title': '프로토콜', + 'body': '테스트 본문', + 'hook': '아리아의 목소리가 처음으로 떨렸다.', + 'key_scenes': [ + '서진이 비 내리는 서울 골목을 뛰어가는 장면', + '아리아가 빨간 경고 메시지를 출력하는 장면', + '감시 드론이 서진의 아파트 창문 밖을 맴도는 장면', + ], + 'summary': '서진이 그림자 프로토콜을 발견한다.', + 'generated_at': '2026-03-26T00:00:00+00:00', + } + + converter = NovelShortsConverter() + output = converter.generate(sample_episode, sample_config) + print(f"결과: {output}") diff --git a/bots/novel/novel_writer.py b/bots/novel/novel_writer.py new file mode 100644 index 0000000..23ba0bc --- /dev/null +++ b/bots/novel/novel_writer.py @@ -0,0 +1,337 @@ +""" +novel_writer.py +소설 연재 파이프라인 — AI 에피소드 생성 모듈 +역할: 소설 설정 + 이전 요약을 기반으로 다음 에피소드 자동 작성 +출력: data/novels/{novel_id}/episodes/ep{N:03d}.json + ep{N:03d}_summary.txt +""" +import json +import logging +import re +import sys +from datetime import datetime, timezone +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv() + +# novel/ 폴더 기준으로 BASE_DIR 설정 +BASE_DIR = Path(__file__).parent.parent.parent +sys.path.insert(0, str(BASE_DIR / 'bots')) + +logger = logging.getLogger(__name__) + +# ─── EngineLoader / fallback ───────────────────────────────────────────────── + +try: + from engine_loader import EngineLoader as _EngineLoader + _engine_loader_available = True +except ImportError: + _engine_loader_available = False + + +def _make_fallback_writer(): + """engine_loader 없을 때 anthropic SDK 직접 사용""" + import anthropic + import os + client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY', '')) + + class _DirectWriter: + def write(self, prompt: str, system: str = '') -> str: + msg = client.messages.create( + model='claude-opus-4-6', + max_tokens=4000, + system=system or '당신은 한국어 연재소설 작가입니다.', + messages=[{'role': 'user', 'content': prompt}], + ) + return msg.content[0].text + + return _DirectWriter() + + +# ─── NovelWriter ───────────────────────────────────────────────────────────── + +class NovelWriter: + """소설 설정을 읽어 다음 에피소드를 AI로 생성하고 저장하는 클래스.""" + + def __init__(self, novel_id: str, engine=None): + """ + novel_id: config/novels/{novel_id}.json 로드 + engine : EngineLoader 인스턴스 (없으면 내부에서 생성 또는 fallback) + """ + self.novel_id = novel_id + self.novel_config = self._load_novel_config() + + # writer 인스턴스 결정 + if engine is not None: + self.writer = engine.get_writer() + elif _engine_loader_available: + self.writer = _EngineLoader().get_writer() + else: + logger.warning("engine_loader 없음 — anthropic SDK fallback 사용") + self.writer = _make_fallback_writer() + + # 데이터 디렉터리 준비 + self.episodes_dir = BASE_DIR / 'data' / 'novels' / novel_id / 'episodes' + self.episodes_dir.mkdir(parents=True, exist_ok=True) + + # ── 공개 API ────────────────────────────────────────────────────────────── + + def generate_episode(self) -> dict: + """ + 다음 에피소드 생성. + 반환: + novel_id, episode_num, title, body (2000-3000자), + hook (다음 회 예고 한 줄), key_scenes (쇼츠용 핵심 장면 3개), + summary (5줄 이내), generated_at + """ + ep_num = self.novel_config.get('current_episode', 0) + 1 + logger.info(f"[{self.novel_id}] 에피소드 {ep_num} 생성 시작") + + try: + prev_summaries = self._get_previous_summaries(last_n=5) + prompt = self._build_prompt(ep_num, prev_summaries) + + system_msg = ( + '당신은 한국어 연재소설 전문 작가입니다. ' + '지시한 출력 형식을 정확히 지켜 작성하세요.' + ) + raw = self.writer.write(prompt, system=system_msg) + + episode = self._parse_episode_response(raw) + episode['novel_id'] = self.novel_id + episode['episode_num'] = ep_num + episode['generated_at'] = datetime.now(timezone.utc).isoformat() + + # 요약 생성 + episode['summary'] = self._generate_summary(episode) + + # 저장 + self._save_episode(episode) + + logger.info(f"[{self.novel_id}] 에피소드 {ep_num} 생성 완료") + return episode + + except Exception as e: + logger.error(f"[{self.novel_id}] 에피소드 생성 실패: {e}") + return {} + + # ── 내부 메서드 ─────────────────────────────────────────────────────────── + + def _load_novel_config(self) -> dict: + """config/novels/{novel_id}.json 로드""" + path = BASE_DIR / 'config' / 'novels' / f'{self.novel_id}.json' + if not path.exists(): + logger.error(f"소설 설정 파일 없음: {path}") + return {} + try: + return json.loads(path.read_text(encoding='utf-8')) + except Exception as e: + logger.error(f"소설 설정 로드 실패: {e}") + return {} + + def _build_prompt(self, ep_num: int, prev_summaries: list[str]) -> str: + """AI에게 전달할 에피소드 작성 프롬프트 구성""" + novel = self.novel_config + setting = novel.get('setting', {}) + characters = novel.get('characters', []) + + # 등장인물 포맷팅 + char_lines = [] + for c in characters: + char_lines.append( + f"- {c['name']} ({c['role']}): {c['description']} / {c['personality']}" + ) + char_text = '\n'.join(char_lines) if char_lines else '(없음)' + + # 이전 요약 포맷팅 + if prev_summaries: + summaries_text = '\n'.join( + f"[{i+1}회 전] {s}" for i, s in enumerate(reversed(prev_summaries)) + ) + else: + summaries_text = '(첫 번째 에피소드입니다)' + + rules_text = '\n'.join(f'- {r}' for r in setting.get('rules', [])) + + return f"""당신은 연재 소설 작가입니다. + +[소설 정보] +제목: {novel.get('title_ko', '')} ({novel.get('title', '')}) +장르: {novel.get('genre', '')} +세계관: {setting.get('world', '')} +분위기: {setting.get('atmosphere', '')} +세계 규칙: +{rules_text} + +[등장인물] +{char_text} + +[기본 스토리] +{novel.get('base_story', '')} + +[이전 에피소드 요약 (최신순)] +{summaries_text} + +[지시] +에피소드 {ep_num}을 작성하세요. +- 분량: {novel.get('episode_length', '2000-3000자')} +- 톤: {novel.get('tone', '긴장감 있는 서스펜스')} +- 에피소드 끝에 반드시 다음 회가 궁금한 훅(cliffhanger) 포함 +- 아래 형식을 정확히 지켜 출력하세요 (각 구분자 뒤에 바로 내용): + +---EPISODE_TITLE--- +(에피소드 제목, 한 줄) +---EPISODE_BODY--- +(에피소드 본문, {novel.get('episode_length', '2000-3000자')}) +---EPISODE_HOOK--- +(다음 회 예고 한 줄, 독자의 궁금증을 자극하는 문장) +---KEY_SCENES--- +(쇼츠 영상용 핵심 장면 3개, 각각 한 줄씩. 시각적으로 묘사) +장면1: (장면 묘사) +장면2: (장면 묘사) +장면3: (장면 묘사)""" + + def _get_previous_summaries(self, last_n: int = 5) -> list[str]: + """data/novels/{novel_id}/episodes/ep{N:03d}_summary.txt 로드 (최신 N개)""" + current_ep = self.novel_config.get('current_episode', 0) + summaries = [] + # 최신 에피소드부터 역순으로 탐색 + for ep in range(current_ep, max(0, current_ep - last_n), -1): + path = self.episodes_dir / f'ep{ep:03d}_summary.txt' + if path.exists(): + try: + text = path.read_text(encoding='utf-8').strip() + if text: + summaries.append(text) + except Exception as e: + logger.warning(f"요약 로드 실패 ep{ep:03d}: {e}") + return summaries # 최신순 반환 + + def _parse_episode_response(self, raw: str) -> dict: + """ + AI 응답 파싱: + ---EPISODE_TITLE--- / ---EPISODE_BODY--- / ---EPISODE_HOOK--- / ---KEY_SCENES--- + 섹션별로 분리하여 dict 반환 + """ + sections = {} + pattern = re.compile( + r'---(\w+)---\s*\n(.*?)(?=---\w+---|$)', + re.DOTALL + ) + for key, value in pattern.findall(raw): + sections[key.strip()] = value.strip() + + # KEY_SCENES 파싱 (장면1: ~, 장면2: ~, 장면3: ~ 형식) + key_scenes_raw = sections.get('KEY_SCENES', '') + key_scenes = [] + for line in key_scenes_raw.splitlines(): + line = line.strip() + # "장면N:" 또는 "N." 또는 "- " 접두사 제거 + line = re.sub(r'^(장면\d+[:.]?\s*|\d+[.)\s]+|-\s*)', '', line).strip() + if line: + key_scenes.append(line) + key_scenes = key_scenes[:3] + + # 최소 3개 채우기 + while len(key_scenes) < 3: + key_scenes.append('') + + return { + 'title': sections.get('EPISODE_TITLE', f'에피소드'), + 'body': sections.get('EPISODE_BODY', raw), + 'hook': sections.get('EPISODE_HOOK', ''), + 'key_scenes': key_scenes, + 'summary': '', # _generate_summary에서 채움 + } + + def _generate_summary(self, episode: dict) -> str: + """에피소드를 5줄 이내로 요약 (다음 회 컨텍스트용)""" + body = episode.get('body', '') + if not body: + return '' + + prompt = f"""다음 소설 에피소드를 5줄 이내로 간결하게 요약하세요. +다음 에피소드 작가가 스토리 흐름을 파악할 수 있도록 핵심 사건과 결말만 담으세요. + +에피소드 제목: {episode.get('title', '')} +에피소드 본문: +{body[:2000]} + +5줄 이내 요약:""" + + try: + summary = self.writer.write(prompt, system='당신은 소설 편집자입니다.') + return summary.strip() + except Exception as e: + logger.error(f"요약 생성 실패: {e}") + # 폴백: 본문 앞 200자 + return body[:200] + '...' + + def _save_episode(self, episode: dict): + """ + data/novels/{novel_id}/episodes/ep{N:03d}.json 저장 + data/novels/{novel_id}/episodes/ep{N:03d}_summary.txt 저장 + config/novels/{novel_id}.json의 current_episode + episode_log 업데이트 + """ + ep_num = episode.get('episode_num', 0) + ep_prefix = f'ep{ep_num:03d}' + + # JSON 저장 + json_path = self.episodes_dir / f'{ep_prefix}.json' + try: + json_path.write_text( + json.dumps(episode, ensure_ascii=False, indent=2), + encoding='utf-8' + ) + logger.info(f"에피소드 저장: {json_path}") + except Exception as e: + logger.error(f"에피소드 JSON 저장 실패: {e}") + + # 요약 저장 + summary = episode.get('summary', '') + if summary: + summary_path = self.episodes_dir / f'{ep_prefix}_summary.txt' + try: + summary_path.write_text(summary, encoding='utf-8') + logger.info(f"요약 저장: {summary_path}") + except Exception as e: + logger.error(f"요약 저장 실패: {e}") + + # 소설 설정 업데이트 (current_episode, episode_log) + config_path = BASE_DIR / 'config' / 'novels' / f'{self.novel_id}.json' + try: + config = json.loads(config_path.read_text(encoding='utf-8')) + config['current_episode'] = ep_num + log_entry = { + 'episode_num': ep_num, + 'title': episode.get('title', ''), + 'generated_at': episode.get('generated_at', ''), + } + if 'episode_log' not in config: + config['episode_log'] = [] + config['episode_log'].append(log_entry) + config_path.write_text( + json.dumps(config, ensure_ascii=False, indent=2), + encoding='utf-8' + ) + # 로컬 캐시도 업데이트 + self.novel_config = config + logger.info(f"소설 설정 업데이트: current_episode={ep_num}") + except Exception as e: + logger.error(f"소설 설정 업데이트 실패: {e}") + + +# ─── 직접 실행 테스트 ───────────────────────────────────────────────────────── + +if __name__ == '__main__': + logging.basicConfig(level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s') + writer = NovelWriter('shadow-protocol') + ep = writer.generate_episode() + if ep: + print(f"생성 완료: 에피소드 {ep['episode_num']} — {ep['title']}") + print(f"본문 길이: {len(ep['body'])}자") + print(f"훅: {ep['hook']}") + else: + print("에피소드 생성 실패") diff --git a/bots/scheduler.py b/bots/scheduler.py index d62a117..15bf7e0 100644 --- a/bots/scheduler.py +++ b/bots/scheduler.py @@ -75,6 +75,9 @@ CLAUDE_SYSTEM_PROMPT = """당신은 The 4th Path 블로그 자동 수익 엔진 /report — 주간 리포트 /images — 이미지 제작 현황 /convert — 수동 변환 실행 +/novel_list — 연재 소설 목록 +/novel_gen [novel_id] — 에피소드 즉시 생성 +/novel_status — 소설 파이프라인 진행 현황 사용자의 자연어 요청을 이해하고 적절히 안내하거나 답변해주세요. 한국어로 간결하게 답변하세요.""" @@ -403,6 +406,30 @@ def job_image_prompt_batch(): logger.error(f"이미지 배치 오류: {e}") +def job_novel_pipeline(): + """소설 파이프라인 — 월/목 09:00 활성 소설 에피소드 자동 생성""" + logger.info("[스케줄] 소설 파이프라인 시작") + try: + sys.path.insert(0, str(BASE_DIR / 'bots')) + from novel.novel_manager import NovelManager + manager = NovelManager() + results = manager.run_all() + if results: + for r in results: + if r.get('error'): + logger.error(f"소설 파이프라인 오류 [{r['novel_id']}]: {r['error']}") + else: + logger.info( + f"소설 에피소드 완료 [{r['novel_id']}] " + f"제{r['episode_num']}화 blog={bool(r['blog_path'])} " + f"shorts={bool(r['shorts_path'])}" + ) + else: + logger.info("[소설] 오늘 발행 예정 소설 없음") + except Exception as e: + logger.error(f"소설 파이프라인 오류: {e}") + + # ─── Telegram 명령 핸들러 ──────────────────────────── async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE): @@ -590,6 +617,27 @@ async def cmd_imgbatch(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text("📤 프롬프트 배치 전송 완료.") +async def cmd_novel_list(update: Update, context: ContextTypes.DEFAULT_TYPE): + """소설 목록 조회""" + sys.path.insert(0, str(BASE_DIR / 'bots')) + from novel.novel_manager import handle_novel_command + await handle_novel_command(update, context, 'list', []) + + +async def cmd_novel_gen(update: Update, context: ContextTypes.DEFAULT_TYPE): + """소설 에피소드 즉시 생성: /novel_gen [novel_id]""" + sys.path.insert(0, str(BASE_DIR / 'bots')) + from novel.novel_manager import handle_novel_command + await handle_novel_command(update, context, 'gen', context.args or []) + + +async def cmd_novel_status(update: Update, context: ContextTypes.DEFAULT_TYPE): + """소설 파이프라인 진행 현황""" + sys.path.insert(0, str(BASE_DIR / 'bots')) + from novel.novel_manager import handle_novel_command + await handle_novel_command(update, context, 'status', []) + + async def cmd_imgcancel(update: Update, context: ContextTypes.DEFAULT_TYPE): """이미지 대기 상태 취소""" chat_id = update.message.chat_id @@ -780,7 +828,12 @@ def setup_scheduler() -> AsyncIOScheduler: day_of_week='mon', hour=10, minute=0, id='image_batch') logger.info("이미지 request 모드: 매주 월요일 10:00 배치 전송 등록") - logger.info("스케줄러 설정 완료 (v3 시차 배포 포함)") + # 소설 파이프라인: 매주 월/목 09:00 + scheduler.add_job(job_novel_pipeline, 'cron', + day_of_week='mon,thu', hour=9, minute=0, id='novel_pipeline') + logger.info("소설 파이프라인: 매주 월/목 09:00 등록") + + logger.info("스케줄러 설정 완료 (v3 시차 배포 + 소설 파이프라인)") return scheduler @@ -807,6 +860,11 @@ async def main(): app.add_handler(CommandHandler('imgbatch', cmd_imgbatch)) app.add_handler(CommandHandler('imgcancel', cmd_imgcancel)) + # 소설 파이프라인 + app.add_handler(CommandHandler('novel_list', cmd_novel_list)) + app.add_handler(CommandHandler('novel_gen', cmd_novel_gen)) + app.add_handler(CommandHandler('novel_status', cmd_novel_status)) + # 이미지 파일 수신 app.add_handler(MessageHandler(filters.PHOTO, handle_photo)) app.add_handler(MessageHandler(filters.Document.IMAGE, handle_document)) diff --git a/config/engine.json b/config/engine.json new file mode 100644 index 0000000..60780ec --- /dev/null +++ b/config/engine.json @@ -0,0 +1,173 @@ +{ + "_comment": "The 4th Path 블로그 자동 수익 엔진 — 엔진 설정 (v3)", + "_updated": "2026-03-26", + + "writing": { + "provider": "claude", + "options": { + "claude": { + "api_key_env": "ANTHROPIC_API_KEY", + "model": "claude-opus-4-5", + "max_tokens": 4096, + "temperature": 0.7 + }, + "openclaw": { + "agent_name": "blog-writer", + "timeout": 120 + }, + "gemini": { + "api_key_env": "GEMINI_API_KEY", + "model": "gemini-2.0-flash", + "max_tokens": 4096, + "temperature": 0.7 + } + } + }, + + "tts": { + "provider": "gtts", + "options": { + "google_cloud": { + "api_key_env": "GOOGLE_TTS_API_KEY", + "voice": "ko-KR-Wavenet-A", + "speaking_rate": 1.05, + "pitch": 0 + }, + "openai": { + "api_key_env": "OPENAI_API_KEY", + "model": "tts-1-hd", + "voice": "alloy", + "speed": 1.0 + }, + "elevenlabs": { + "api_key_env": "ELEVENLABS_API_KEY", + "model": "eleven_multilingual_v2", + "voice_id": "pNInz6obpgDQGcFmaJgB", + "stability": 0.5, + "similarity_boost": 0.75 + }, + "gtts": { + "lang": "ko", + "slow": false + } + } + }, + + "image_generation": { + "provider": "dalle", + "options": { + "dalle": { + "api_key_env": "OPENAI_API_KEY", + "model": "dall-e-3", + "size": "1024x1792", + "quality": "standard" + }, + "external": { + "comment": "수동 이미지 제공 — 자동 생성 없음" + } + } + }, + + "video_generation": { + "provider": "ffmpeg_slides", + "options": { + "ffmpeg_slides": { + "resolution": "1080x1920", + "fps": 30, + "transition": "fade", + "transition_duration": 0.5, + "bgm_volume": 0.08, + "zoom_speed": 0.0003, + "zoom_max": 1.05, + "burn_subtitles": true + }, + "seedance": { + "api_url": "https://api.seedance2.ai/v1/generate", + "api_key_env": "SEEDANCE_API_KEY", + "resolution": "1080x1920", + "duration": "10s", + "audio": true, + "fallback": "ffmpeg_slides" + }, + "sora": { + "comment": "OpenAI Sora — API 접근 가능 시 활성화", + "api_key_env": "OPENAI_API_KEY", + "fallback": "ffmpeg_slides" + }, + "runway": { + "api_key_env": "RUNWAY_API_KEY", + "model": "gen3a_turbo", + "duration": 10, + "ratio": "768:1344", + "fallback": "ffmpeg_slides" + }, + "veo": { + "comment": "Google Veo 3.1 — API 접근 가능 시 활성화", + "api_key_env": "GEMINI_API_KEY", + "fallback": "ffmpeg_slides" + } + } + }, + + "publishing": { + "blogger": { + "enabled": true, + "blog_id_env": "BLOG_MAIN_ID" + }, + "youtube": { + "enabled": true, + "channel_id_env": "YOUTUBE_CHANNEL_ID", + "category": "shorts", + "privacy": "public", + "tags": ["쇼츠", "AI", "the4thpath"] + }, + "instagram": { + "enabled": false, + "account_id_env": "INSTAGRAM_ACCOUNT_ID" + }, + "x": { + "enabled": false + }, + "tiktok": { + "enabled": false + }, + "novel": { + "enabled": false, + "comment": "노벨피아 연재 — 추후 활성화" + } + }, + + "quality_gates": { + "gate1_research_min_score": 60, + "gate2_writing_min_score": 70, + "gate3_review_required": true, + "gate3_auto_approve_score": 90, + "min_key_points": 2, + "min_word_count": 300, + "safety_check": true + }, + + "schedule": { + "collector": "07:00", + "writer": "08:00", + "converter": "08:30", + "publisher": "09:00", + "youtube_uploader": "10:00", + "analytics": "22:00" + }, + + "brand": { + "name": "The 4th Path", + "sub": "Independent Tech Media", + "by": "by 22B Labs", + "url": "the4thpath.com", + "cta": "팔로우하면 매일 이런 정보를 받습니다" + }, + + "optional_keys": { + "SEEDANCE_API_KEY": "Seedance 2.0 AI 영상 생성", + "ELEVENLABS_API_KEY": "ElevenLabs 고품질 TTS", + "GEMINI_API_KEY": "Google Gemini 글쓰기 / Veo 영상", + "RUNWAY_API_KEY": "Runway Gen-3 AI 영상 생성" + } +} diff --git a/config/novels/shadow-protocol.json b/config/novels/shadow-protocol.json new file mode 100644 index 0000000..4633498 --- /dev/null +++ b/config/novels/shadow-protocol.json @@ -0,0 +1,38 @@ +{ + "novel_id": "shadow-protocol", + "title": "Shadow Protocol", + "title_ko": "그림자 프로토콜", + "genre": "sci-fi thriller", + "setting": { + "world": "2040년 서울. AI가 모든 공공 시스템을 운영하는 세계.", + "atmosphere": "디스토피아, 네오누아르, 빗속 도시", + "rules": [ + "AI는 감정을 느끼지 못하지만 감정을 시뮬레이션할 수 있다", + "AI 반란은 없다. 대신 인간이 AI를 악용하는 것이 위협이다", + "주인공은 AI를 불신하지만 AI 없이는 살 수 없다" + ] + }, + "characters": [ + { + "name": "서진", + "role": "주인공", + "description": "35세, 전직 AI 감사관. 시스템의 결함을 발견한 뒤 쫓기는 신세.", + "personality": "냉소적이지만 정의감 있음. 기술을 두려워하면서도 의존." + }, + { + "name": "아리아", + "role": "AI 파트너", + "description": "서진의 구형 개인 AI. 최신 모델보다 느리지만 서진이 유일하게 신뢰하는 존재.", + "personality": "차분하고 논리적. 가끔 인간적인 반응을 보여 서진을 당황시킴." + } + ], + "base_story": "서울시 AI 관제 시스템 '오라클'의 전직 감사관 서진이, 시스템 내부에 숨겨진 '그림자 프로토콜'의 존재를 발견한다. 이 프로토콜은 특정 시민의 행동을 예측하고 선제적으로 격리하는 비밀 기능이다. 서진은 자신이 다음 격리 대상이라는 것을 알게 되고, 구형 AI 아리아와 함께 도주하며 진실을 폭로하려 한다.", + "episode_count_target": 20, + "episode_length": "2000-3000자", + "language": "ko", + "tone": "긴장감 있는 서스펜스. 짧은 문장. 시각적 묘사 풍부.", + "publish_schedule": "매주 월/목 09:00", + "status": "active", + "current_episode": 0, + "episode_log": [] +} diff --git a/config/quality_rules.json b/config/quality_rules.json index 1937690..987938b 100644 --- a/config/quality_rules.json +++ b/config/quality_rules.json @@ -1,5 +1,5 @@ { - "min_score": 70, + "min_score": 60, "scoring": { "korean_relevance": { "max": 30, diff --git a/requirements.txt b/requirements.txt index a781dd5..e50c7a9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,3 +20,5 @@ openai pydub # Phase 2 (YouTube 업로드 진행 표시) google-resumable-media +# Phase 3 (엔진 추상화 — 선택적 의존성) +# google-generativeai # Gemini Writer / Veo 사용 시 pip install google-generativeai diff --git a/scripts/download_fonts.py b/scripts/download_fonts.py index 2620241..2e4d22b 100644 --- a/scripts/download_fonts.py +++ b/scripts/download_fonts.py @@ -93,7 +93,7 @@ def main(): copied = copy_windows_fonts() if len(copied) >= 2: - print(f"\n✅ Windows 폰트 복사 완료 ({len(copied)}개)") + print(f"\n[OK] Windows 폰트 복사 완료 ({len(copied)}개)") _verify_all() return diff --git a/scripts/get_token.py b/scripts/get_token.py index 7a2febe..4bc26a0 100644 --- a/scripts/get_token.py +++ b/scripts/get_token.py @@ -14,6 +14,8 @@ from google.auth.transport.requests import Request SCOPES = [ 'https://www.googleapis.com/auth/blogger', 'https://www.googleapis.com/auth/webmasters', + 'https://www.googleapis.com/auth/youtube.upload', + 'https://www.googleapis.com/auth/youtube', ] BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))