feat: v3.0 엔진 추상화 + 소설 파이프라인 추가

[1순위] 엔진 추상화 리팩토링
- config/engine.json: 단일 설정 파일로 writing/tts/image/video/publishing 엔진 제어
- bots/engine_loader.py: EngineLoader 팩토리 클래스 (Claude/OpenClaw/Gemini Writer, gTTS/GoogleCloud/OpenAI/ElevenLabs TTS, DALL-E/External 이미지)

[2순위] VideoEngine 추상화
- bots/converters/video_engine.py: VideoEngine ABC + FFmpegSlidesEngine/SeedanceEngine/SoraEngine/RunwayEngine/VeoEngine 구현
- Seedance 2.0 API 연동 + 실패 시 ffmpeg_slides 자동 fallback

[3순위] 소설 연재 파이프라인
- bots/novel/novel_writer.py: AI 에피소드 자동 생성 (Claude/엔진 추상화)
- bots/novel/novel_blog_converter.py: 에피소드 → 장르별 테마 Blogger HTML
- bots/novel/novel_shorts_converter.py: key_scenes → TTS + Pillow + VideoEngine → MP4
- bots/novel/novel_manager.py: 전체 파이프라인 조율 + Telegram 명령 처리
- config/novels/shadow-protocol.json: 예시 소설 설정 (2040 서울 SF 스릴러)

[스케줄러] 소설 파이프라인 통합
- 매주 월/목 09:00 자동 실행 (job_novel_pipeline)
- Telegram 명령: /novel_list, /novel_gen, /novel_status

[기타 수정]
- collector_bot.py: 한국어 유니코드 감지 + RSS 신뢰도 override 버그 수정
- quality_rules.json: min_score 70→60
- scripts/get_token.py: YouTube OAuth scope 추가
- .env.example: SEEDANCE/ELEVENLABS/GEMINI/RUNWAY API 키 항목 추가

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
sinmb79
2026-03-26 09:33:04 +09:00
parent b54f8e198e
commit 8a7a122bb3
16 changed files with 3487 additions and 8 deletions

View File

@@ -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=

View File

@@ -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,8 +229,13 @@ 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)
# 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)

View File

@@ -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)

521
bots/engine_loader.py Normal file
View File

@@ -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)]

0
bots/novel/__init__.py Normal file
View File

View File

@@ -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 (
'<script type="application/ld+json">\n'
+ json.dumps(schema, ensure_ascii=False, indent=2)
+ '\n</script>'
)
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('&', '&amp;')
.replace('<', '&lt;')
.replace('>', '&gt;'))
# 대화문 (따옴표 시작) 스타일 적용
if line.startswith('"') or line.startswith('"') or line.startswith('"'):
html_lines.append(
f'<span style="color:inherit;opacity:0.9">{line}</span>'
)
else:
html_lines.append(line)
paragraphs.append('<br>\n'.join(html_lines))
return '\n'.join(
f'<p style="margin:0 0 1.6em 0;line-height:1.9;letter-spacing:0.01em">{p}</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'<a href="{prev_url}" style="color:{accent};text-decoration:none;font-size:14px">&#8592; {ep_num - 1}화</a>'
if prev_url and ep_num > 1
else f'<span style="color:{meta_color};font-size:14px">첫 번째 에피소드</span>'
)
next_link = (
f'<a href="{next_url}" style="color:{accent};text-decoration:none;font-size:14px">{ep_num + 1}화 &#8594;</a>'
if next_url
else f'<span style="color:{meta_color};font-size:14px">다음 회 업데이트 예정</span>'
)
# 전체 HTML 조립
html = f"""{json_ld}
<style>
.post-title {{ display:none!important }}
</style>
<div style="max-width:680px;margin:0 auto;padding:24px 16px;background:{bg};font-family:-apple-system,BlinkMacSystemFont,'Segoe UI','Malgun Gothic','Apple SD Gothic Neo',sans-serif;color:{text_color}">
<!-- 에피소드 배지 -->
<div style="margin-bottom:20px">
<span style="display:inline-block;background:{accent};color:#fff;font-size:12px;font-weight:700;letter-spacing:0.08em;padding:5px 14px;border-radius:20px;text-transform:uppercase">
연재소설 · 에피소드 {ep_num}
</span>
</div>
<!-- 소설 제목 -->
<p style="margin:0 0 6px 0;font-size:13px;color:{meta_color};letter-spacing:0.05em;font-weight:600">
{title_ko}
</p>
<!-- 에피소드 제목 -->
<h1 style="margin:0 0 20px 0;font-size:28px;font-weight:800;line-height:1.3;color:#fff;letter-spacing:-0.01em">
{title}
</h1>
<!-- 메타 정보 -->
<div style="display:flex;align-items:center;gap:12px;margin-bottom:32px;padding-bottom:20px;border-bottom:1px solid {accent_dim}40">
<span style="font-size:13px;color:{meta_color}">{genre}</span>
<span style="color:{accent_dim}">·</span>
<span style="font-size:13px;color:{meta_color}">{novel_config.get('episode_length','')}</span>
<span style="color:{accent_dim}">·</span>
<span style="font-size:13px;color:{meta_color}">{datetime.now().strftime('%Y.%m.%d')}</span>
</div>
<!-- 에피소드 이전/다음 네비게이션 (상단) -->
<div style="display:flex;justify-content:space-between;align-items:center;margin-bottom:32px;padding:14px 18px;background:{nav_bg};border-radius:10px;border:1px solid {accent_dim}30">
{prev_link}
<span style="font-size:12px;color:{meta_color};font-weight:600">{ep_num}화</span>
{next_link}
</div>
<!-- 본문 -->
<div style="font-size:17px;line-height:1.9;color:{text_color}">
{body_html}
</div>
<!-- 구분선 -->
<div style="margin:40px 0 32px 0;height:2px;background:linear-gradient(to right,{accent},{accent}00)"></div>
<!-- 클로징 박스: 다음 에피소드 예고 -->
<div style="background:{card_bg};border-left:4px solid {accent};border-radius:0 12px 12px 0;padding:20px 22px;margin-bottom:32px">
<p style="margin:0 0 8px 0;font-size:12px;font-weight:700;color:{accent};letter-spacing:0.08em;text-transform:uppercase">
다음 에피소드 예고 · {next_date_str}
</p>
<p style="margin:0;font-size:15px;line-height:1.7;color:{text_color}">
{hook if hook else '다음 회를 기대해 주세요.'}
</p>
</div>
<!-- AdSense 슬롯 -->
<!-- AD_SLOT_NOVEL -->
<!-- 에피소드 이전/다음 네비게이션 (하단) -->
<div style="display:flex;justify-content:space-between;align-items:center;margin-top:32px;padding:14px 18px;background:{nav_bg};border-radius:10px;border:1px solid {accent_dim}30">
{prev_link}
<a href="#" style="color:{meta_color};text-decoration:none;font-size:13px">&#8593; 목록</a>
{next_link}
</div>
<!-- 소설 정보 푸터 -->
<div style="margin-top:32px;padding:20px;background:{card_bg};border-radius:12px;border:1px solid {accent_dim}20">
<p style="margin:0 0 8px 0;font-size:13px;font-weight:700;color:{accent}">
{title_ko} 정보
</p>
<p style="margin:0 0 6px 0;font-size:13px;color:{meta_color}">
장르: {genre} · 목표 {novel_config.get('episode_count_target', 20)}화 완결
</p>
<p style="margin:0;font-size:12px;color:{meta_color}">
연재 일정: {novel_config.get('publish_schedule', '')} · The 4th Path
</p>
</div>
</div>"""
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)}")

507
bots/novel/novel_manager.py Normal file
View File

@@ -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))

View File

@@ -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}")

337
bots/novel/novel_writer.py Normal file
View File

@@ -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("에피소드 생성 실패")

View File

@@ -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))

173
config/engine.json Normal file
View File

@@ -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 영상 생성"
}
}

View File

@@ -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": []
}

View File

@@ -1,5 +1,5 @@
{
"min_score": 70,
"min_score": 60,
"scoring": {
"korean_relevance": {
"max": 30,

View File

@@ -20,3 +20,5 @@ openai
pydub
# Phase 2 (YouTube 업로드 진행 표시)
google-resumable-media
# Phase 3 (엔진 추상화 — 선택적 의존성)
# google-generativeai # Gemini Writer / Veo 사용 시 pip install google-generativeai

View File

@@ -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

View File

@@ -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__)))