# Autoworker Pipeline Upgrade Design: Kling AI 비디오 클립 통합

> 작성일: 2026-03-27
> 목적: Flux 정지이미지 + Ken Burns → Kling AI 실제 비디오 클립으로 고도화

---

## 1. 현황 분석

### 현재 파이프라인 흐름 (Stage 3-B)

```
storyboard_hints
    └─ visual 필드 (한국어 묘사)
           │
           ▼
    HF Flux API → PNG 이미지 생성
           │         (실패 시 Pillow 솔리드 폴백)
           ▼
    ffmpeg zoompan → Ken Burns MP4 클립
           │
           ▼
    assembly.py → 최종 영상
```

### 핵심 문제

| 문제 | 원인 | 영향 |
|------|------|------|
| 영상 퀄리티 낮음 | 정지이미지 + Ken Burns 패닝 효과 | 시청자 이탈 증가 |
| Kling 사용 불가 | video_prompt 없음, 한국어 묘사만 존재 | AI 비디오 생성 불가 |
| 비용 비효율 | 씬 우선순위 구분 없음 | 불필요한 크레딧 소모 |

---

## 2. 목표 파이프라인 아키텍처

```
storyboard_hints
    ├─ visual (한국어, 참고용)
    ├─ video_prompt (영어, Kling 입력)   ← NEW
    ├─ clip_priority: high/low           ← NEW
    └─ duration_sec (오디오 실측값)

           │
           ▼
    [clip_priority 라우터]
           │
     ┌─────┴─────┐
     │           │
  high          low
     │           │
     ▼           ▼
  Kling API   HF Flux API
  (fal.ai)    → Ken Burns
     │           │
     └─────┬─────┘
           │
           ▼
    씬별 MP4 클립
           │
           ▼
    assembly.py → 최종 영상
```

---

## 3. fal.ai Python SDK 레퍼런스

### 설치

```bash
pip install fal-client
```

### 인증

```bash
export FAL_KEY="your-fal-api-key"
```

또는 코드에서:
```python
import os
os.environ["FAL_KEY"] = "your-fal-api-key"
import fal_client
```

### 핵심 메서드

#### 방법 1: `subscribe()` — 가장 간단 (블로킹, 자동 폴링)

```python
import fal_client

result = fal_client.subscribe(
    "fal-ai/kling-video/v2.1/standard/text-to-video",
    arguments={
        "prompt": "Cinematic shot of Korean office workers in 1997",
        "duration": "5",           # "5" 또는 "10"
        "aspect_ratio": "16:9",
        "negative_prompt": "text, watermark, cartoon, blur, distorted",
    },
    with_logs=False,
    on_queue_update=lambda update: print(f"Status: {update}"),
)
# result["video"]["url"] → 다운로드 URL
```

#### 방법 2: `submit()` / `result()` — 비동기 병렬 처리

```python
import fal_client

# 제출 (즉시 반환)
handler = fal_client.submit(
    "fal-ai/kling-video/v2.1/standard/text-to-video",
    arguments={
        "prompt": "...",
        "duration": "5",
        "aspect_ratio": "16:9",
        "negative_prompt": "text, watermark, cartoon, blur, distorted",
    }
)
request_id = handler.request_id  # 나중에 결과 조회용

# 결과 조회 (블로킹)
result = handler.get()
video_url = result["video"]["url"]
```

#### 방법 3: `submit_async()` — 완전 비동기 (병렬 씬 생성 권장)

```python
import asyncio
import fal_client

async def generate_clip_async(prompt: str, duration: int) -> str:
    handler = await fal_client.submit_async(
        "fal-ai/kling-video/v2.1/standard/text-to-video",
        arguments={
            "prompt": prompt,
            "duration": str(duration),
            "aspect_ratio": "16:9",
            "negative_prompt": "text, watermark, cartoon, blur, distorted",
        }
    )
    result = await handler.get()
    return result["video"]["url"]

async def generate_all_clips(scenes: list[dict]) -> list[str]:
    tasks = [generate_clip_async(s["video_prompt"], s["duration"]) for s in scenes]
    return await asyncio.gather(*tasks)  # 모든 씬 병렬 생성
```

### Queue 상태 라이프사이클

```
IN_QUEUE → IN_PROGRESS → COMPLETED
```

- **IN_QUEUE**: 러너 대기 중
- **IN_PROGRESS**: 생성 중 (30초~2분)
- **COMPLETED**: 결과 URL 사용 가능

---

## 4. visual.py 수정 설계

### 전체 구조

```python
"""
Stage 3-B: VISUAL - 하이브리드 비디오 클립 생성
Kling AI (fal.ai) 실제 비디오 → Flux 정지이미지 + Ken Burns 폴백
"""

import asyncio
import logging
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import requests

logger = logging.getLogger(__name__)
```

### 핵심 함수: `_generate_clip_kling()`

```python
def _generate_clip_kling(
    prompt: str,
    out_path: Path,
    duration: int,          # 5 또는 10 (Kling 지원값)
    settings: dict,
    style_cfg: dict = None, # settings["visual_style"]
) -> bool:
    """
    fal.ai Kling API로 실제 비디오 클립 생성.
    
    Args:
        prompt: 영어 비디오 프롬프트 (storyboard_hints의 video_prompt)
        out_path: 출력 MP4 경로
        duration: 클립 길이 (5 또는 10초만 지원)
        settings: 채널 settings.json
        style_cfg: visual_style 설정 (prefix, negative)
    
    Returns:
        True: 성공, False: 실패 (폴백으로 전환)
    """
    fal_key = os.environ.get("FAL_KEY")
    if not fal_key:
        logger.warning("FAL_KEY 없음. Kling 생성 스킵.")
        return False

    # fal_client import (런타임 의존성)
    try:
        import fal_client
    except ImportError:
        logger.warning("fal-client 미설치: pip install fal-client")
        return False

    # 프롬프트 보강 (visual_style prefix 적용)
    style = style_cfg or settings.get("visual_style", {})
    prefix = style.get("prefix", "cinematic, photorealistic, 4K")
    negative = style.get("negative", "text, watermark, cartoon, blur")
    
    full_prompt = f"{prefix}, {prompt}" if prefix else prompt

    # Kling duration은 5 또는 10만 허용
    kling_duration = 10 if duration >= 8 else 5

    video_cfg = settings.get("video_generation", {})
    model = video_cfg.get("model", "fal-ai/kling-video/v2.1/standard/text-to-video")
    aspect_ratio = video_cfg.get("aspect_ratio", "16:9")

    try:
        logger.info(f"Kling 생성 시작: duration={kling_duration}s, prompt='{full_prompt[:60]}...'")
        t0 = time.time()
        
        result = fal_client.subscribe(
            model,
            arguments={
                "prompt": full_prompt,
                "duration": str(kling_duration),
                "aspect_ratio": aspect_ratio,
                "negative_prompt": negative,
            },
            with_logs=False,
            on_queue_update=lambda u: logger.debug(f"Kling status: {u}"),
        )
        
        elapsed = time.time() - t0
        video_url = result.get("video", {}).get("url") or result.get("url")
        
        if not video_url:
            logger.warning(f"Kling: video URL 없음. result keys: {list(result.keys())}")
            return False

        # 다운로드
        resp = requests.get(video_url, timeout=120, stream=True)
        resp.raise_for_status()
        out_path.write_bytes(resp.content)
        
        size_kb = len(resp.content) // 1024
        logger.info(f"Kling 성공: {out_path.name} ({size_kb}KB, {elapsed:.1f}s)")
        return True

    except Exception as e:
        logger.warning(f"Kling API 실패: {e}")
        return False
```

### 씬별 병렬 생성 함수

```python
def _generate_clips_parallel(
    scenes: list[dict],      # [{"idx": 0, "prompt": "...", "duration": 5, "priority": "high"}, ...]
    clips_dir: Path,
    settings: dict,
    max_workers: int = 3,    # 동시 API 호출 수 (fal.ai 제한 고려)
) -> dict[int, str]:
    """
    씬 목록을 병렬로 Kling 생성.
    
    Returns:
        {scene_idx: clip_path_str} — 성공한 씬만 포함
    """
    results = {}
    
    def generate_one(scene: dict) -> tuple[int, str | None]:
        idx = scene["idx"]
        clip_path = clips_dir / f"scene_{idx:02d}.mp4"
        success = _generate_clip_kling(
            prompt=scene["prompt"],
            out_path=clip_path,
            duration=scene["duration"],
            settings=settings,
        )
        return (idx, str(clip_path) if success else None)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(generate_one, s): s for s in scenes}
        for future in as_completed(futures):
            idx, path = future.result()
            if path:
                results[idx] = path
    
    return results
```

### 메인 함수: `generate_visuals()` (수정 버전)

```python
def generate_visuals(
    storyboard_hints: list[dict],
    project_dir: Path,
    settings: dict,
) -> dict:
    """
    Stage 3-B: 하이브리드 비디오 클립 생성.
    
    storyboard_hints 항목 (수정 후):
    {
        "index": int,
        "prompt": str,            # 이미지 생성용 (Flux 폴백)
        "video_prompt": str,      # NEW: 영어 비디오 프롬프트 (Kling용)
        "duration": int,
        "emotion": str,
        "clip_priority": str,     # NEW: "high" (Kling) | "low" (Flux)
    }
    """
    images_dir = project_dir / "visual" / "images"
    clips_dir = project_dir / "visual" / "clips"
    images_dir.mkdir(parents=True, exist_ok=True)
    clips_dir.mkdir(parents=True, exist_ok=True)

    clips: list[str] = []
    images: list[str] = []

    # 오디오 실측 길이 로드 (기존 로직 유지)
    audio_durations = _load_audio_durations(project_dir / "audio")
    default_duration = settings.get("video", {}).get("default_scene_duration", 8)

    video_cfg = settings.get("video_generation", {})
    budget_per_video = video_cfg.get("budget_credits_per_video", 500)
    used_credits = 0

    # 씬 분류: high priority (Kling) vs low priority (Flux)
    kling_scenes = []
    flux_scenes = []
    
    for i, hint in enumerate(storyboard_hints):
        idx = hint.get("index", i)
        duration = audio_durations.get(idx) or hint.get("duration") or default_duration
        duration = max(1, int(round(duration)))
        
        priority = hint.get("clip_priority", "low")
        video_prompt = hint.get("video_prompt") or hint.get("prompt", f"Scene {idx}")
        
        scene = {
            "idx": idx,
            "hint": hint,
            "prompt": hint.get("prompt", f"Scene {idx}"),
            "video_prompt": video_prompt,
            "duration": duration,
            "priority": priority,
        }
        
        if priority == "high" and video_prompt:
            # 크레딧 예산 체크
            est_credits = _estimate_kling_credits(duration)
            if used_credits + est_credits <= budget_per_video:
                kling_scenes.append(scene)
                used_credits += est_credits
            else:
                logger.warning(f"씬 {idx}: 예산 초과 → Flux 폴백 (used={used_credits}/{budget_per_video})")
                flux_scenes.append(scene)
        else:
            flux_scenes.append(scene)
    
    logger.info(f"씬 분류: Kling={len(kling_scenes)}, Flux={len(flux_scenes)}, 예상크레딧={used_credits}")

    # === Kling 병렬 생성 ===
    kling_results = {}
    if kling_scenes:
        max_workers = video_cfg.get("parallel_workers", 3)
        kling_results = _generate_clips_parallel(kling_scenes, clips_dir, settings, max_workers)
        logger.info(f"Kling 생성 완료: {len(kling_results)}/{len(kling_scenes)}개 성공")

    # === Flux + Ken Burns (기존 로직) ===
    # Kling 실패한 씬도 Flux로 폴백
    failed_kling = [s for s in kling_scenes if s["idx"] not in kling_results]
    all_flux_scenes = flux_scenes + failed_kling
    
    for scene in all_flux_scenes:
        idx = scene["idx"]
        img_path = images_dir / f"scene_{idx:02d}.png"
        clip_path = clips_dir / f"scene_{idx:02d}.mp4"
        
        success = _generate_image_hf_flux(scene["prompt"], img_path, settings)
        if not success:
            _fallback_image(idx, img_path)
        images.append(str(img_path))
        
        effect = _pick_effect(idx, scene["hint"])
        try:
            _ken_burns_clip(img_path, clip_path, duration=scene["duration"], effect=effect)
        except Exception:
            _simple_loop_clip(img_path, clip_path, duration=scene["duration"])

    # === 최종 클립 순서 조합 ===
    for i, hint in enumerate(storyboard_hints):
        idx = hint.get("index", i)
        clip_path = clips_dir / f"scene_{idx:02d}.mp4"
        if clip_path.exists():
            clips.append(str(clip_path))

    logger.info(f"비주얼 완료: Kling {len(kling_results)}개, Flux {len(all_flux_scenes)}개")
    return {"clips": clips, "images": images}


def _estimate_kling_credits(duration: int) -> int:
    """Kling 씬 하나당 예상 크레딧 소비."""
    # fal.ai Kling v2.1 Standard 기준 (2026-03 기준):
    # 5초: ~35 크레딧, 10초: ~70 크레딧
    # fal.ai 크레딧 가격: $1 = 100 크레딧
    return 70 if duration >= 8 else 35


def _load_audio_durations(audio_dir: Path) -> dict[int, float]:
    """오디오 실측 길이 로드 (기존 로직)."""
    import subprocess as _sp
    result = {}
    if not audio_dir.exists():
        return result
    for mp3_file in audio_dir.glob("scene_*.mp3"):
        try:
            res = _sp.run(
                ["ffprobe", "-v", "quiet", "-show_entries", "format=duration",
                 "-of", "csv=p=0", str(mp3_file)],
                capture_output=True, text=True
            )
            dur = float(res.stdout.strip())
            scene_idx = int(mp3_file.stem.split("_")[1])
            result[scene_idx] = dur
        except Exception:
            pass
    return result
```

---

## 5. script.py 수정 설계

### 프롬프트 수정: `_build_script_prompt()` 내 storyboard_hints 섹션

현재 JSON 출력 스펙:
```json
{"scene_id": 1, "visual": "한국어 장면 묘사", "emotion": "serious", "duration_sec": 30}
```

수정 후 storyboard_hints 출력 스펙:

```python
# _build_script_prompt() 내 OUTPUT JSON 섹션 수정
STORYBOARD_HINTS_SPEC = """
  "storyboard_hints": [
    {
      "scene_id": 1,
      "visual": "화면에 보여줄 것 (한국어 묘사, 참고용)",
      "video_prompt": "Cinematic shot of [description in English], [camera movement], [lighting style], [color tone]",
      "emotion": "neutral | excited | serious | curious | warm | dramatic",
      "duration_sec": 5,
      "clip_priority": "high | low"
    }
  ]
"""
```

### clip_priority 판단 기준 (Claude 프롬프트에 지시사항 추가)

```
clip_priority 판정 규칙:
- "high" (Kling 생성, 실제 비디오 클립):
  * 훅 씬 (처음 30초): 시청자 이탈 방지 최우선
  * 클라이맥스 씬: 감정적으로 가장 중요한 장면
  * 인물/행동 씬: 사람이 직접 등장하거나 중요한 행동
  * 브랜드 강조 씬: 채널 아이덴티티와 직결된 장면
  
- "low" (Flux 정지이미지 + Ken Burns):
  * 전환/간격 씬: 단순 배경이나 공간 연출
  * 텍스트 오버레이 씬: 데이터, 통계 표시 위주
  * 반복 배경 씬: 짧은 설명 보조 영상
  * 아웃트로/인트로 로고 씬
```

### 업데이트된 프롬프트 조각

```python
# _build_script_prompt()에 추가할 섹션
VIDEO_PROMPT_INSTRUCTIONS = """
STORYBOARD VIDEO PROMPTS (for each scene in storyboard_hints):

1. video_prompt MUST be in English, optimized for AI video generation
2. Structure: "[shot type], [subject description], [action/movement], [style], [lighting], [color grade]"
   Examples:
   - "Close-up of exhausted Korean businessman, 1997 Seoul office, fluorescent lighting, archival film grain, warm amber tones, slow zoom in"
   - "Aerial shot of Seoul cityscape at night, 1997 era buildings, bokeh city lights, cinematic wide angle, slow pan left"
   - "Medium shot of protest crowd in Seoul street, 1997, documentary style, high contrast, handheld camera movement"

3. clip_priority = "high" for: hooks, climaxes, character moments, brand-defining scenes
   clip_priority = "low" for: transitions, background B-roll, data overlay scenes, outro

4. video_prompt length: 15-50 words (optimal for Kling AI)
5. NEVER include text, subtitles, or watermarks in video_prompt
"""
```

---

## 6. 하이브리드 전략: 크레딧 최적화

### Kling 크레딧 비용 구조 (fal.ai 기준)

```
모델: fal-ai/kling-video/v2.1/standard/text-to-video
5초 클립:  ~$0.35 (≈35 크레딧)
10초 클립: ~$0.70 (≈70 크레딧)

v2.1 pro (고화질):
5초 클립:  ~$0.52 (≈52 크레딧)
10초 클립: ~$1.05 (≈105 크레딧)
```

### 영상 1편 기준 예산 시나리오

```
일반 10분 유튜브 (약 12-15 씬):
  - 전체 Kling v2.1 standard (5초): 15 × $0.35 = $5.25
  - 전체 Kling v2.1 standard (10초): 15 × $0.70 = $10.50

하이브리드 (high 씬 40% = 6개 Kling, 나머지 Flux):
  - 6 × $0.35 (5초) = $2.10 → Flux 대비 퀄리티 핵심 씬만 업그레이드
  - 권장: budget_credits_per_video = 250 (~$2.50)

월 20편 제작 기준:
  - 전체 Kling: 20 × $5.25 = $105/월
  - 하이브리드:  20 × $2.10 = $42/월 ← 권장 시작점
  - fal.ai starter plan ($10/mo): 약 4-5편 전체 Kling 가능
```

### clip_priority 자동 판단 로직

```python
def _auto_clip_priority(scene_id: int, total_scenes: int, emotion: str, visual_hint: str) -> str:
    """
    씬 특성 기반 clip_priority 자동 판단.
    (script.py에서 Claude 프롬프트로 결정하는 것이 주 방법,
     이 함수는 폴백 또는 후처리 검증용)
    """
    # 위치 기반
    if scene_id <= 2:  # 훅 (처음 2씬)
        return "high"
    if scene_id >= total_scenes - 1:  # 아웃트로
        return "low"
    if scene_id == total_scenes // 2:  # 중간 클라이맥스
        return "high"
    
    # 감정 기반
    HIGH_EMOTIONS = {"dramatic", "exciting", "emotional", "intense", "climax"}
    LOW_EMOTIONS = {"neutral", "calm", "background"}
    
    if emotion.lower() in HIGH_EMOTIONS:
        return "high"
    if emotion.lower() in LOW_EMOTIONS:
        return "low"
    
    # 키워드 기반 (한국어 visual 필드 분석)
    HIGH_KEYWORDS = ["인물", "등장", "대화", "충격", "결말", "반전", "클라이맥스"]
    LOW_KEYWORDS = ["배경", "풍경", "전환", "통계", "그래프", "자막"]
    
    for kw in HIGH_KEYWORDS:
        if kw in visual_hint:
            return "high"
    for kw in LOW_KEYWORDS:
        if kw in visual_hint:
            return "low"
    
    # 기본값: 예산 절약
    return "low"
```

---

## 7. settings.json 업데이트

```json
{
  "channel": {
    "id": "",
    "name": "my-channel",
    "youtube_channel_id": "UCxxxxxx"
  },
  "content": {
    "type": "explainer",
    "target_duration_min": 10,
    "style_guide": "informative yet engaging. Use data and examples.",
    "target_audience": "general audience interested in the topic",
    "language": "ko"
  },
  "pipeline": {
    "variant": "B",
    "parallel_stage3": true,
    "human_review_script": false,
    "resume_on_failure": true
  },
  "tts": {
    "provider": "elevenlabs",
    "voice_id": "ksaI0TCD9BstzEzlxj4q",
    "model": "eleven_multilingual_v2",
    "stability": 0.4,
    "similarity_boost": 0.85,
    "fallback_provider": "edge-tts",
    "fallback_voice": "ko-KR-InJoonNeural"
  },
  "image": {
    "provider": "pillow_solid",
    "model": "none",
    "style_preset": "photorealistic",
    "fallback_provider": "pillow_solid",
    "gemini_disabled": true
  },
  "video_generation": {
    "provider": "kling",
    "fallback": "flux",
    "model": "fal-ai/kling-video/v2.1/standard/text-to-video",
    "max_clips_per_video": 15,
    "clip_duration": 5,
    "aspect_ratio": "16:9",
    "budget_credits_per_video": 250,
    "parallel_workers": 3,
    "timeout_sec": 180
  },
  "visual_style": {
    "prefix": "cinematic, photorealistic, 4K, professional lighting",
    "negative": "text, watermark, cartoon, blur, distorted, watermark, logo, subtitle"
  },
  "audio": {
    "bgm_enabled": true,
    "bgm_volume": 0.12
  },
  "publish": {
    "privacy": "public",
    "schedule_enabled": false,
    "schedule_time": "09:00",
    "schedule_timezone": "Asia/Seoul"
  }
}
```

---

## 8. 구현 체크리스트

### Phase 1: 기반 작업 (1-2시간)

- [ ] `pip install fal-client` → requirements.txt에 추가
- [ ] `FAL_KEY` → `~/.zshenv`에 등록
- [ ] settings.json에 `video_generation`, `visual_style` 섹션 추가
- [ ] fal.ai 계정에서 Kling v2.1 API 접근 확인

### Phase 2: script.py 수정 (30분)

- [ ] `_build_script_prompt()` 내 OUTPUT JSON에 `video_prompt`, `clip_priority` 필드 추가
- [ ] VIDEO_PROMPT_INSTRUCTIONS 지시사항 프롬프트에 추가
- [ ] `_parse_script_response()` — 새 필드 포함 검증 로직 추가
- [ ] `save_script_artifacts()` — storyboard_hints에 새 필드 포함 확인

### Phase 3: visual.py 수정 (2-3시간)

- [ ] `_generate_clip_kling()` 함수 구현
- [ ] `_generate_clips_parallel()` 구현 (ThreadPoolExecutor)
- [ ] `_estimate_kling_credits()` 구현
- [ ] `generate_visuals()` 메인 함수 — 하이브리드 라우터 추가
- [ ] Kling 실패 시 Flux 자동 폴백 로직

### Phase 4: 테스트 (1시간)

- [ ] 단일 씬 Kling 생성 테스트
- [ ] 예산 초과 시 Flux 폴백 동작 확인
- [ ] 병렬 3개 씬 동시 생성 테스트
- [ ] 전체 파이프라인 E2E 테스트 (test-channel)

---

## 9. 주의사항 및 리스크

### API 제한

```
fal.ai 동시 요청: 계정 플랜에 따라 다름 (보통 3-5개)
→ parallel_workers = 3 권장 (안전 기본값)

Kling 생성 시간: 30초~3분 (씬 복잡도에 따라)
→ timeout_sec = 180 설정 필수
→ 전체 15씬 병렬 처리 시 최대 ~3분 소요 예상
```

### 비용 안전장치

```python
# visual.py 내 하드 리밋
MAX_CREDITS_HARD_LIMIT = 1000  # 영상 1편당 절대 한도

if used_credits >= MAX_CREDITS_HARD_LIMIT:
    logger.error(f"크레딧 하드 리밋 도달 ({MAX_CREDITS_HARD_LIMIT}). 나머지 씬 Flux 폴백.")
    break
```

### 클립 길이 미스매치 처리

```python
# Kling은 5초 또는 10초 고정 출력 → 오디오와 길이 차이 발생 가능
# assembly.py에서 ffmpeg -t 옵션으로 오디오 길이에 맞게 트리밍/루프 처리 필요
# 예: 씬 오디오 = 7초, Kling 클립 = 5초 → 루프 or 마지막 프레임 freeze

# assembly.py 관련 메모:
# 현재 assembly.py가 클립 길이를 어떻게 처리하는지 확인 후
# setpts 필터 또는 -stream_loop 옵션 적용 필요
```

---

## 10. 완성된 아키텍처 다이어그램

```
┌─────────────────────────────────────────────────────────────┐
│                   autoworker Pipeline v2                     │
│                  (Kling 하이브리드 통합)                      │
└─────────────────────────────────────────────────────────────┘

Stage 1: research.py ──→ research_data, references
         │
Stage 2: script.py ──→ full_script + storyboard_hints
         │              ├─ visual (KO, 참고용)
         │              ├─ video_prompt (EN, Kling용)  ← NEW
         │              ├─ clip_priority: high/low     ← NEW
         │              └─ duration_sec (오디오 실측)
         │
Stage 3-A: audio.py ──→ scene_XX.mp3 (오디오 실측 길이 확정)
         │
Stage 3-B: visual.py ──→ [라우터]
         │                  │
         │          ┌───────┴───────┐
         │       high씬            low씬
         │     (Kling API)        (Flux API)
         │          │                │
         │    fal_client.subscribe   HF Inference
         │    (병렬 3개 동시)         │
         │    submit → poll          Ken Burns
         │    → download             zoompan
         │          │                │
         │    실패 시 Flux 폴백 ──────┘
         │          │
         │    scene_XX.mp4 (통합)
         │
Stage 4: assembly.py ──→ ffmpeg concat → raw_video.mp4
         │
Stage 5: packaging.py ──→ thumbnail + metadata
         │
Stage 6: publish.py ──→ YouTube upload
```

---

## 11. 참고: fal.ai 크레딧 관리

```python
# FAL_KEY 잔액 확인 (공식 API 미지원, 대시보드에서 확인)
# https://fal.ai/dashboard → Billing → Credits

# 실제 소비 로깅 패턴
class KlingCreditTracker:
    def __init__(self, budget: int):
        self.budget = budget
        self.used = 0
    
    def can_afford(self, duration: int) -> bool:
        cost = 70 if duration >= 8 else 35
        return self.used + cost <= self.budget
    
    def consume(self, duration: int):
        cost = 70 if duration >= 8 else 35
        self.used += cost
        logger.info(f"크레딧 사용: +{cost} (총 {self.used}/{self.budget})")
```

---

*이 설계 문서는 autoworker 파이프라인의 Kling AI 통합을 위한 완전한 구현 가이드입니다.
실제 구현 전 fal.ai 대시보드에서 Kling v2.1 모델 접근 권한 및 크레딧 잔액을 확인하세요.*