콘텐츠 자동화 파이프라인 구축 가이드

전체 아키텍처 개요

[Cron/스케줄러] → [매크로 수집 봇] → [Discord 알림]
        ↓
[인턴 에이전트] → 주제 정렬 & 중복 제거
        ↓
[트레이더 에이전트] → 투자 추천 → [Discord 전송]
        ↓
[Claude API] → 번역 → [Notion 업로드]
        ↓
[사용자] → 10개 내외 자료 선택
        ↓
[ElevenLabs API] → TTS 생성
[이미지 합성] → 칠판 배경 합성
        ↓
[JSON 저장] → 작업 데이터 관리
        ↓
[Python 숏츠 생성] → 이미지 + TTS + 번역 → 영상
        ↓
[사용자 승인] → YouTube 업로드
        ↓
[Playwright] → 텔레그램, 네이버블로그, 링크드인, 페이스북, X 자동 게시

---

Phase 1. 인프라 세팅

1-1. 서버 환경

항목	권장 사양
서버	AWS EC2 (t3.medium 이상) 또는 개인 서버
OS	Ubuntu 22.04 LTS
Python	3.11+
Node.js	18+ (Playwright용)
스케줄러	Cron (Linux) 또는 APScheduler (Python)
DB	SQLite (경량) 또는 PostgreSQL

1-2. 필수 계정 & API 키

서비스	용도	비용
Discord Bot	알림 & 결과 전송	무료
Claude API (Anthropic)	번역 & AI 에이전트	종량제
ElevenLabs API	TTS 음성 생성	무료 티어 + 유료
Notion API	데이터 업로드 & 관리	무료
YouTube Data API	영상 업로드	무료
Telegram Bot API	텔레그램 포스팅	무료
Naver Blog API	블로그 포스팅	무료 (또는 Playwright)

1-3. 프로젝트 폴더 구조

content-pipeline/
├── config/
│   ├── .env                  # API 키, 토큰
│   └── settings.yaml         # 스케줄, 채널ID 등 설정
├── collectors/
│   ├── macro_collector.py    # 매크로 데이터 수집
│   └── sources.yaml          # 수집 소스 목록
├── agents/
│   ├── intern_agent.py       # 주제 정렬 & 중복 제거
│   └── trader_agent.py       # 투자 추천 분석
├── translators/
│   └── claude_translator.py  # Claude API 번역
├── media/
│   ├── tts_generator.py      # ElevenLabs TTS
│   ├── image_composer.py     # 칠판 배경 이미지 합성
│   └── shorts_maker.py       # 숏츠 영상 생성
├── publishers/
│   ├── youtube_uploader.py   # YouTube 업로드
│   ├── discord_bot.py        # Discord 알림
│   ├── notion_uploader.py    # Notion 업로드
│   └── social_poster.py      # Playwright 멀티플랫폼 게시
├── data/
│   ├── raw/                  # 수집 원본
│   ├── processed/            # 정렬·중복제거 후
│   ├── selected/             # 사용자 선택분
│   └── output.json           # 최종 작업 데이터
├── assets/
│   ├── chalkboard_bg.png     # 칠판 배경 이미지
│   └── fonts/                # 폰트 파일
├── scheduler.py              # 메인 스케줄러
├── requirements.txt
└── README.md

---

Phase 2. 단계별 구현

Step 1. 매크로 자료 수집 (매일 오후 정해진 시간)

역할: 정해진 소스에서 매크로/금융 데이터를 자동 수집

# scheduler.py — APScheduler 예시
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('cron', hour=14, minute=0)  # 매일 오후 2시
def run_pipeline():
    collect_data()      # Step 1
    notify_discord()    # Step 2
    sort_and_dedup()    # Step 3
    analyze_trading()   # Step 4
    translate_upload()  # Step 5
    # Step 6~12는 사용자 입력 대기

scheduler.start()

수집 소스 예시:

• RSS 피드 (Bloomberg, Reuters 등)
• 웹 스크래핑 (Selenium/Playwright)
• API (금융 데이터 API, 뉴스 API)

필요 라이브러리:

feedparser        # RSS 파싱
requests          # HTTP 요청
beautifulsoup4    # HTML 파싱
playwright        # 동적 페이지 스크래핑

---

Step 2. Discord 알림 — “업무 시작”

역할: 파이프라인 시작을 Discord 채널에 알림

# discord_bot.py
import discord
from discord import Webhook
import aiohttp

WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR_WEBHOOK"

async def notify_start(collected_count: int):
    async with aiohttp.ClientSession() as session:
        webhook = Webhook.from_url(WEBHOOK_URL, session=session)
        embed = discord.Embed(
            title="📡 매크로 수집 완료",
            description=f"총 {collected_count}건의 자료를 수집했습니다.\n인턴 에이전트가 정리를 시작합니다.",
            color=0x00ff00
        )
        await webhook.send(embed=embed)

Tip: Discord Bot 대신 Webhook을 사용하면 더 간단합니다.

---

Step 3. 인턴 에이전트 — 주제 정렬 & 중복 제거

역할: Claude API를 활용하여 수집된 자료를 주제별 분류하고, 유사 콘텐츠 제거

# intern_agent.py
import anthropic

client = anthropic.Anthropic(api_key="YOUR_KEY")

def sort_and_deduplicate(articles: list[dict]) -> list[dict]:
    prompt = f"""
    다음 {len(articles)}개의 매크로 자료를 분석해주세요:
    
    1. 주제별로 분류 (예: 금리, 환율, 원자재, 주식, 부동산 등)
    2. 내용이 유사한 자료는 가장 정보량이 많은 것만 남기고 제거
    3. 각 자료에 중요도 점수(1~10) 부여
    
    자료 목록:
    {json.dumps(articles, ensure_ascii=False)}
    
    JSON 형식으로 반환해주세요.
    """
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(response.content[0].text)

---

Step 4. 트레이더 에이전트 — 투자 추천 → Discord 전송

역할: 정리된 자료를 바탕으로 투자 인사이트를 생성하고 Discord에 전송

# trader_agent.py
def generate_trading_insight(sorted_articles: list[dict]) -> str:
    prompt = f"""
    당신은 전문 매크로 트레이더입니다.
    아래 정리된 매크로 자료를 바탕으로:
    
    1. 현재 시장 상황 요약 (3줄)
    2. 주목할 섹터/자산 TOP 3
    3. 각 추천에 대한 근거
    4. 리스크 요인
    
    자료: {json.dumps(sorted_articles, ensure_ascii=False)}
    """
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

# Discord로 전송
async def send_trading_insight(insight: str):
    embed = discord.Embed(
        title="📊 트레이더 에이전트 분석",
        description=insight,
        color=0xffd700
    )
    await webhook.send(embed=embed)

---

Step 5. Claude API 번역 → Notion 업로드

역할: 영문 자료를 한국어로 번역하고 Notion 데이터베이스에 업로드

# claude_translator.py
def translate_article(article: dict) -> dict:
    prompt = f"""
    아래 영문 기사를 자연스러운 한국어로 번역해주세요.
    전문 용어는 원문을 괄호에 병기합니다.
    
    제목: {article['title']}
    본문: {article['content']}
    """
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}]
    )
    article['translated'] = response.content[0].text
    return article

# notion_uploader.py
from notion_client import Client

notion = Client(auth="YOUR_NOTION_TOKEN")
DATABASE_ID = "YOUR_DATABASE_ID"

def upload_to_notion(article: dict):
    notion.pages.create(
        parent={"database_id": DATABASE_ID},
        properties={
            "제목": {"title": [{"text": {"content": article['title_kr']}}]},
            "카테고리": {"select": {"name": article['category']}},
            "중요도": {"number": article['importance']},
            "날짜": {"date": {"start": article['date']}},
            "상태": {"select": {"name": "대기"}}
        },
        children=[
            {"paragraph": {"rich_text": [{"text": {"content": article['translated']}}]}}
        ]
    )

---

Step 6. 사용자 자료 선택 (수동)

역할: Notion 또는 Discord에서 10개 내외 자료를 선택

구현 옵션:

방식	장점	단점
Notion 체크박스	직관적, 모바일 가능	Notion API 폴링 필요
Discord 리액션	빠름, 알림	UI 제한적
웹 대시보드	커스텀 UI	추가 개발 필요

# Notion 기반 선택 감지 예시
def get_selected_articles() -> list[dict]:
    results = notion.databases.query(
        database_id=DATABASE_ID,
        filter={"property": "상태", "select": {"equals": "선택됨"}}
    )
    return [parse_page(page) for page in results['results']]

---

Step 7. ElevenLabs TTS 변환

역할: 선택된 자료의 번역문을 음성으로 변환

# tts_generator.py
from elevenlabs import ElevenLabs

client = ElevenLabs(api_key="YOUR_KEY")

def generate_tts(text: str, output_path: str, voice_id: str = "YOUR_VOICE_ID"):
    audio = client.text_to_speech.convert(
        voice_id=voice_id,
        text=text,
        model_id="eleven_multilingual_v2",
        output_format="mp3_44100_128"
    )
    
    with open(output_path, "wb") as f:
        for chunk in audio:
            f.write(chunk)
    
    return output_path

음성 설정 팁:

• eleven_multilingual_v2 모델 → 한국어 지원
• Voice Cloning으로 본인 목소리 사용 가능
• Stability: 0.5, Similarity: 0.75 기본 권장

---

Step 8. 이미지 칠판 배경 합성

역할: 핵심 내용을 칠판 스타일 이미지로 합성

# image_composer.py
from PIL import Image, ImageDraw, ImageFont

def compose_chalkboard_image(
    title: str,
    key_points: list[str],
    bg_path: str = "assets/chalkboard_bg.png",
    output_path: str = "output.png"
):
    bg = Image.open(bg_path).resize((1080, 1920))  # 숏츠 세로 비율
    draw = ImageDraw.Draw(bg)
    
    # 폰트 설정 (분필 느낌)
    title_font = ImageFont.truetype("assets/fonts/chalk_font.ttf", 60)
    body_font = ImageFont.truetype("assets/fonts/chalk_font.ttf", 40)
    
    # 제목
    draw.text((540, 200), title, font=title_font, fill="white", anchor="mt")
    
    # 핵심 포인트
    y = 400
    for i, point in enumerate(key_points):
        draw.text((100, y), f"✦ {point}", font=body_font, fill="#E8E8E8")
        y += 80
    
    bg.save(output_path)
    return output_path

---

Step 9. JSON 저장

역할: 모든 작업 데이터를 JSON으로 저장하여 숏츠 생성에 활용

# data/output.json 구조
{
    "date": "2026-03-15",
    "articles": [
        {
            "id": 1,
            "title_en": "Fed Holds Rates Steady",
            "title_kr": "연준, 금리 동결 결정",
            "category": "금리",
            "summary_kr": "연준이 3월 FOMC에서...",
            "translated_full": "...",
            "importance": 9,
            "tts_path": "data/selected/01_tts.mp3",
            "image_path": "data/selected/01_chalkboard.png",
            "duration_seconds": 45
        }
    ],
    "trading_insight": "...",
    "total_duration": 540
}

---

Step 10. Python 숏츠 생성

역할: 이미지 + TTS + 자막을 조합하여 YouTube Shorts 영상 생성

# shorts_maker.py
from moviepy.editor import (
    ImageClip, AudioFileClip, TextClip, 
    CompositeVideoClip, concatenate_videoclips
)

def create_shorts(data_json: str, output_path: str = "output_shorts.mp4"):
    with open(data_json) as f:
        data = json.load(f)
    
    clips = []
    for article in data['articles']:
        # 이미지 클립
        audio = AudioFileClip(article['tts_path'])
        image = ImageClip(article['image_path']).set_duration(audio.duration)
        
        # 자막 추가
        subtitle = TextClip(
            article['summary_kr'],
            fontsize=30, color='white', bg_color='rgba(0,0,0,0.7)',
            size=(1000, None), method='caption'
        ).set_position(('center', 1600)).set_duration(audio.duration)
        
        # 합성
        clip = CompositeVideoClip([image, subtitle]).set_audio(audio)
        clips.append(clip)
    
    # 인트로/아웃트로 추가 가능
    final = concatenate_videoclips(clips, method="compose")
    final.write_videofile(
        output_path,
        fps=30,
        codec='libx264',
        audio_codec='aac'
    )
    return output_path

필요 라이브러리:

moviepy          # 영상 편집
Pillow           # 이미지 처리
ffmpeg           # 코덱 (시스템 설치 필요)

---

Step 11. 사용자 승인 → YouTube 업로드

역할: Discord에서 [승인] 버튼 클릭 시 YouTube에 업로드

# Discord 버튼 인터랙션
class ApprovalView(discord.ui.View):
    @discord.ui.button(label="✅ 승인", style=discord.ButtonStyle.green)
    async def approve(self, interaction, button):
        await interaction.response.send_message("📤 업로드를 시작합니다...")
        upload_to_youtube(shorts_path)
        await interaction.followup.send("✅ YouTube 업로드 완료!")
    
    @discord.ui.button(label="❌ 반려", style=discord.ButtonStyle.red)
    async def reject(self, interaction, button):
        await interaction.response.send_message("🔄 수정이 필요합니다. 피드백을 입력해주세요.")

# YouTube 업로드
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials

def upload_to_youtube(video_path: str, title: str, description: str):
    creds = Credentials.from_authorized_user_file('config/youtube_token.json')
    youtube = build('youtube', 'v3', credentials=creds)
    
    request = youtube.videos().insert(
        part="snippet,status",
        body={
            "snippet": {
                "title": title,
                "description": description,
                "tags": ["매크로", "경제", "투자"],
                "categoryId": "22"  # People & Blogs
            },
            "status": {
                "privacyStatus": "public",
                "selfDeclaredMadeForKids": False
            }
        },
        media_body=video_path
    )
    response = request.execute()
    return response['id']

---

Step 12. Playwright 멀티 플랫폼 자동 게시

역할: 숏츠 업로드 후 텍스트 버전을 여러 플랫폼에 자동 게시

# social_poster.py
from playwright.async_api import async_playwright

class SocialPoster:
    async def post_all(self, content: dict):
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            
            await self.post_telegram(browser, content)
            await self.post_naver_blog(browser, content)
            await self.post_linkedin(browser, content)
            await self.post_facebook(browser, content)
            await self.post_x(browser, content)
            
            await browser.close()
    
    async def post_telegram(self, browser, content):
        # Telegram Bot API 사용 (Playwright 불필요)
        import httpx
        BOT_TOKEN = "YOUR_BOT_TOKEN"
        CHAT_ID = "YOUR_CHAT_ID"
        url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
        await httpx.AsyncClient().post(url, json={
            "chat_id": CHAT_ID,
            "text": content['text'],
            "parse_mode": "Markdown"
        })
    
    async def post_naver_blog(self, browser, content):
        page = await browser.new_page()
        await page.goto("https://blog.naver.com")
        # 로그인 → 글쓰기 → 내용 입력 → 발행
        # (세부 셀렉터는 네이버 블로그 UI에 맞게 조정 필요)
        await page.close()
    
    async def post_linkedin(self, browser, content):
        page = await browser.new_page()
        await page.goto("https://www.linkedin.com")
        # 로그인 → 게시물 작성 → 발행
        await page.close()
    
    async def post_facebook(self, browser, content):
        page = await browser.new_page()
        await page.goto("https://www.facebook.com")
        # 로그인 → 게시물 작성 → 발행
        await page.close()
    
    async def post_x(self, browser, content):
        page = await browser.new_page()
        await page.goto("https://x.com")
        # 로그인 → 트윗 작성 → 발행
        await page.close()

⚠️ 주의: Playwright 기반 SNS 자동 게시는 각 플랫폼의 이용약관에 따라 계정 제한 리스크가 있습니다. 가능하면 공식 API를 우선 사용하세요.

---

Phase 3. 추가 자동화 (🟥 확장)

유튜브 영상 기반 주제 추천

# topic_recommender.py
def recommend_topics_from_youtube(channel_id: str):
    """기존 유튜브 영상을 분석하여 새로운 주제를 추천"""
    # 1. YouTube API로 채널 영상 목록 가져오기
    # 2. 영상 제목/설명/태그 수집
    # 3. Claude API로 트렌드 분석 & 새 주제 추천
    
    prompt = f"""
    다음은 유튜브 채널의 최근 영상 목록입니다:
    {video_list}
    
    1. 어떤 주제가 조회수가 높았는지 분석
    2. 아직 다루지 않은 관련 주제 5개 추천
    3. 현재 트렌드와 연결되는 주제 3개 추천
    """
    return claude_response

전체 스케줄 관리

# config/settings.yaml
schedules:
  macro_collection:
    time: "14:00"
    timezone: "Asia/Seoul"
    enabled: true
  
  topic_recommendation:
    time: "09:00"
    frequency: "weekly"  # 매주 월요일
    enabled: true
  
  analytics_report:
    time: "10:00"
    frequency: "weekly"  # 매주 월요일
    enabled: true
  
  channel_performance:
    time: "08:00"
    frequency: "daily"
    enabled: false

---

기술 스택 요약

카테고리	기술	용도
언어	Python 3.11+	전체 파이프라인
스케줄링	APScheduler / Cron	정시 실행
AI	Claude API (Anthropic)	번역, 분류, 분석, 추천
TTS	ElevenLabs API	음성 생성
영상	MoviePy + FFmpeg	숏츠 제작
이미지	Pillow	칠판 배경 합성
알림	Discord.py / Webhook	상태 알림 & 승인
데이터	Notion API	자료 관리
자동화	Playwright	SNS 자동 게시
업로드	YouTube Data API v3	영상 업로드
서버	AWS EC2	24시간 운영

---

필요 패키지 (requirements.txt)

# AI & NLP
anthropic>=0.40.0
elevenlabs>=1.0.0

# 데이터 수집
feedparser
requests
beautifulsoup4

# 영상 & 이미지
moviepy>=2.0.0
Pillow>=10.0.0

# 자동화
playwright
apscheduler>=3.10.0

# Discord
discord.py>=2.3.0
aiohttp

# Notion
notion-client>=2.0.0

# YouTube
google-api-python-client
google-auth-oauthlib

# SNS
httpx

# 유틸
python-dotenv
pyyaml

---

구현 순서 (권장)

한 번에 전체를 만들지 말고, 아래 순서로 하나씩 완성하세요.

Phase 1 — MVP (1~2주)
├── ① 매크로 수집 봇 (collectors/)
├── ② Discord 알림 (publishers/discord_bot.py)
├── ③ 인턴 에이전트 (agents/intern_agent.py)
└── ④ Notion 업로드 (publishers/notion_uploader.py)

Phase 2 — 콘텐츠 생성 (2~3주)
├── ⑤ Claude 번역 (translators/)
├── ⑥ TTS 생성 (media/tts_generator.py)
├── ⑦ 이미지 합성 (media/image_composer.py)
└── ⑧ 숏츠 생성 (media/shorts_maker.py)

Phase 3 — 배포 자동화 (1~2주)
├── ⑨ YouTube 업로드 (publishers/youtube_uploader.py)
├── ⑩ 승인 워크플로우 (Discord 버튼)
└── ⑪ 멀티플랫폼 게시 (publishers/social_poster.py)

Phase 4 — 고도화 (지속)
├── ⑫ 유튜브 분석 & 주제 추천
├── ⑬ 성과 대시보드
└── ⑭ A/B 테스트 (썸네일, 제목 등)

---

주의사항

[!CAUTION]

• API 비용 관리: Claude API, ElevenLabs는 종량제입니다. 일일 사용량 상한을 설정하세요.
• SNS 자동화 리스크: Playwright 기반 자동 게시는 계정 정지 리스크가 있습니다. 공식 API 우선 사용을 권장합니다.
• 환경변수 보안: .env 파일은 절대 Git에 커밋하지 마세요. .gitignore에 반드시 추가하세요.
• 영상 저작권: 수집한 자료의 출처를 반드시 명시하고, 저작권 문제가 없는 콘텐츠만 사용하세요.