diff --git a/bots/config_resolver.py b/bots/config_resolver.py new file mode 100644 index 0000000..e03f4e1 --- /dev/null +++ b/bots/config_resolver.py @@ -0,0 +1,232 @@ +""" +bots/config_resolver.py [NEW] + +Single source of truth at runtime. +Merges user_profile + engine.json + env. + +Priority: user_profile > engine.json > hardcoded defaults +Missing API key → auto-downgrade to free alternative +""" + +import json +import os +import sys +from pathlib import Path + +# Base directory of the project (one level up from bots/) +BASE_DIR = Path(__file__).resolve().parent.parent + +# Fallback engine for each category when all else fails +FALLBACKS = { + 'writing': 'openclaw', + 'tts': 'edge_tts', + 'video': 'ffmpeg_slides', + 'image': 'external', +} + +# Budget-to-engine priority lists per category +BUDGET_ENGINE_MAP = { + 'free': { + 'writing': ['openclaw', 'claude_web', 'gemini_web'], + 'tts': ['kokoro', 'edge_tts'], + 'video': ['kling_free', 'ffmpeg_slides'], + 'image': ['external'], + }, + 'low': { + 'writing': ['openclaw', 'claude_web', 'claude'], + 'tts': ['openai_tts', 'kokoro', 'edge_tts'], + 'video': ['kling_free', 'veo3', 'seedance2', 'ffmpeg_slides'], + 'image': ['dalle', 'external'], + }, + 'medium': { + 'writing': ['openclaw', 'claude', 'gemini'], + 'tts': ['elevenlabs', 'openai_tts', 'cosyvoice2', 'edge_tts'], + 'video': ['kling_free', 'veo3', 'seedance2', 'runway', 'ffmpeg_slides'], + 'image': ['dalle', 'external'], + }, + 'premium': { + 'writing': ['openclaw', 'claude', 'gemini'], + 'tts': ['elevenlabs', 'openai_tts', 'cosyvoice2'], + 'video': ['kling_free', 'veo3', 'seedance2', 'runway', 'kling_pro'], + 'image': ['dalle', 'midjourney', 'external'], + }, +} + +# Engine registry: local=True means no API key required (free/local) +ENGINE_REGISTRY = { + 'kokoro': {'local': True}, + 'edge_tts': {'local': True}, + 'ffmpeg_slides': {'local': True}, + 'external': {'local': True}, + 'cosyvoice2': {'local': True}, + 'openclaw': {'local': True}, + 'claude_web': {'local': True}, + 'gemini_web': {'local': True}, + # API-based engines + 'elevenlabs': {'local': False}, + 'openai_tts': {'local': False}, + 'claude': {'local': False}, + 'gemini': {'local': False}, + 'kling_free': {'local': False}, + 'kling_pro': {'local': False}, + 'veo3': {'local': False}, + 'seedance2': {'local': False}, + 'runway': {'local': False}, + 'dalle': {'local': False}, + 'midjourney': {'local': False}, +} + +# Map from engine name to required environment variable +ENGINE_API_KEY_MAP = { + 'elevenlabs': 'ELEVENLABS_API_KEY', + 'openai_tts': 'OPENAI_API_KEY', + 'claude': 'ANTHROPIC_API_KEY', + 'gemini': 'GEMINI_API_KEY', + 'kling_free': 'KLING_API_KEY', + 'kling_pro': 'KLING_API_KEY', + 'veo3': 'GEMINI_API_KEY', + 'seedance2': 'FAL_API_KEY', + 'runway': 'RUNWAY_API_KEY', + 'dalle': 'OPENAI_API_KEY', + 'midjourney': 'MIDJOURNEY_API_KEY', +} + + +class ConfigResolver: + """ + Single source of truth at runtime. + Merges user_profile + engine.json + env. + + Priority: user_profile > engine.json > hardcoded defaults + Missing API key → auto-downgrade to free alternative + """ + + def resolve(self) -> dict: + """Resolve and return the full runtime configuration.""" + profile = self._load('config/user_profile.json') + engine = self._load('config/engine.json') + + resolved = { + 'writing': self._resolve_engine('writing', profile, engine), + 'tts': self._resolve_engine('tts', profile, engine), + 'video': self._resolve_engine('video', profile, engine), + 'image': self._resolve_engine('image', profile, engine), + 'platforms': self._resolve_platforms(profile), + 'budget': profile.get('budget', 'free'), + 'level': profile.get('level', 'beginner'), + } + return resolved + + def _load(self, path: str) -> dict: + """Load JSON from BASE_DIR/path; return {} if file not found or invalid.""" + full_path = BASE_DIR / path + try: + with open(full_path, encoding='utf-8') as f: + return json.load(f) + except FileNotFoundError: + print(f"[설정] {path} 없음 — 기본값 사용", file=sys.stderr) + return {} + except json.JSONDecodeError as e: + print(f"[설정] {path} 파싱 오류: {e} — 기본값 사용", file=sys.stderr) + return {} + + def _has_api_key(self, engine_name: str) -> bool: + """ + Check whether the required API key env var for the given engine is set. + Engines not in ENGINE_API_KEY_MAP are local/free and always available. + """ + # Local/free engines never need a key + engine_info = ENGINE_REGISTRY.get(engine_name, {}) + if engine_info.get('local', False): + return True + + env_var = ENGINE_API_KEY_MAP.get(engine_name) + if env_var is None: + # Unknown engine — treat as available (graceful degradation) + return True + + value = os.environ.get(env_var, '').strip() + return len(value) > 0 + + def _resolve_engine(self, category: str, profile: dict, engine: dict) -> dict: + """ + Resolve the active engine for a category. + + Steps: + 1. Check user's chosen provider from profile + 2. Check if that provider's API key exists in env + 3. If not, auto-switch to next available alternative within budget + 4. If all fail, use hardcoded free fallback + + Returns dict with 'provider' and 'auto_selected' flag. + """ + budget = profile.get('budget', 'free') + candidate_list = BUDGET_ENGINE_MAP.get(budget, BUDGET_ENGINE_MAP['free']).get(category, []) + + # Determine user's preferred provider + engines_section = profile.get('engines', {}) + category_cfg = engines_section.get(category, {}) + user_provider = category_cfg.get('provider', 'auto') if isinstance(category_cfg, dict) else 'auto' + + # If user explicitly set a provider (not "auto"), try it first + if user_provider and user_provider != 'auto': + if self._has_api_key(user_provider): + print(f"[설정] {category}: 사용자 지정 '{user_provider}' 사용") + return {'provider': user_provider, 'auto_selected': False} + else: + print(f"[설정] {category}: '{user_provider}' API 키 없음 — 자동 선택으로 전환") + + # Auto-select: iterate budget-appropriate candidates in priority order + for engine_name in candidate_list: + if self._has_api_key(engine_name): + auto = (user_provider == 'auto') + if not auto: + print(f"[설정] {category}: '{engine_name}'으로 자동 전환") + else: + print(f"[설정] {category}: 자동 선택 → '{engine_name}'") + return {'provider': engine_name, 'auto_selected': True} + + # Last resort: hardcoded free fallback + fallback = FALLBACKS.get(category, 'external') + print(f"[설정] {category}: 모든 엔진 실패 — 기본 폴백 '{fallback}' 사용") + return {'provider': fallback, 'auto_selected': True} + + def _resolve_platforms(self, profile: dict) -> list: + """Return the list of target publishing platforms from user profile.""" + return profile.get('platforms', []) + + +# --------------------------------------------------------------------------- +# Standalone test entry point +# --------------------------------------------------------------------------- + +def _run_test(): + """Print resolved config for manual verification.""" + print("=" * 60) + print("ConfigResolver 테스트 실행") + print("=" * 60) + + resolver = ConfigResolver() + config = resolver.resolve() + + print("\n[결과] 런타임 설정:") + print(json.dumps(config, ensure_ascii=False, indent=2)) + + print("\n[요약]") + print(f" 예산 등급 : {config['budget']}") + print(f" 사용자 레벨: {config['level']}") + print(f" 플랫폼 : {config['platforms']}") + for cat in ('writing', 'tts', 'video', 'image'): + eng = config[cat] + flag = '(자동)' if eng.get('auto_selected') else '(지정)' + print(f" {cat:10s}: {eng['provider']} {flag}") + + print("=" * 60) + print("테스트 완료") + + +if __name__ == '__main__': + if '--test' in sys.argv: + _run_test() + else: + print("사용법: python -m bots.config_resolver --test") diff --git a/config/user_profile.json b/config/user_profile.json new file mode 100644 index 0000000..6114874 --- /dev/null +++ b/config/user_profile.json @@ -0,0 +1,18 @@ +{ + "_comment": "사용자 의도 설정 — bw init으로 생성/업데이트", + "_updated": "2026-03-29", + "budget": "free", + "level": "beginner", + "engines": { + "writing": {"provider": "auto"}, + "tts": {"provider": "auto"}, + "video": {"provider": "auto"}, + "image": {"provider": "auto"} + }, + "platforms": ["youtube"], + "services": { + "openclaw": false, + "claude_web": false, + "gemini_web": false + } +}