YouTube 자동화 & 업무 효율화 시스템

AI 기반 영상 요약 및 개인 업무 자동화 통합 플랫폼

📋 프로젝트 개요

개인의 반복적인 업무를 자동화하여 효율성을 극대화하기 위한 통합 플랫폼입니다. YouTube 영상 자동 업로드를 시작으로, AI 기반 영상 요약, 근태 관리 자동화, 학습 데이터 수집까지 확장된 개인용 업무 지원 시스템입니다.

핵심 가치

• 시간 절약: 반복 작업 자동화로 일일 2시간 절감
• 콘텐츠 재활용: 긴 영상을 자막 + 요약으로 빠르게 파악
• 실수 방지: 근태 체크 자동화로 누락 0건

🎯 해결한 문제

반복 작업의 비효율

1. 수동 영상 업로드

   • 매일 생성되는 강의/회의 영상을 수동 업로드
   • 제목, 설명, 태그 매번 입력
   • 업로드 진행률 확인 불가

2. 콘텐츠 재확인 시간 낭비

   • 긴 영상을 다시 보며 내용 확인
   • 핵심 내용 파악에 시간 소요
   • 자막 없어서 검색 어려움

3. 근태 관리 누락

   • 출퇴근 체크 잊어버림
   • 증빙 자료 수동 생성
   • 관리 시스템 매번 접속

4. 학습 데이터 정리

   • LMS 다시보기 수동 확인
   • 목록 정리 반복 작업

통합 자동화 솔루션

• YouTube API: 메타데이터와 함께 자동 업로드
• AI 파이프라인: Whisper STT → Gemini/GLM 요약
• 근태 자동화: API 연동으로 자동 체크
• 실시간 UI: Socket.IO로 진행률 표시

🛠 기술 스택

Backend

• Framework: Flask 2.3.3
• Real-time: Flask-SocketIO
• Process Management: PM2 (ecosystem.config.js)

AI/ML Pipeline

• STT: OpenAI Whisper (Custom API)
• LLM:
  • Google Gemini Pro/Flash (주 엔진)
  • Z.ai GLM-4 (대안 엔진)
• Audio: FFmpeg (영상 → 음성 추출)

External APIs

• YouTube: YouTube Data API v3
• Notification: Slack Webhook
• Attendance: 업무 시스템 REST API

Infrastructure

• Platform: Windows 10/11
• Storage: 로컬 파일 시스템 (JSON, TXT, MD)
• Session: Flask Session

💡 주요 구현 내용

1. YouTube 자동 업로드 시스템

파일 감시 + 웹 UI

# youtube_uploader.py
class YouTubeUploader:
    def __init__(self, client_secret_path):
        self.youtube = self._get_authenticated_service(
            client_secret_path
        )
    
    def upload_video(self, file_path, metadata, 
                     progress_callback=None):
        """
        청크 업로드 + 실시간 진행률
        """
        media = MediaFileUpload(
            file_path,
            chunksize=10*1024*1024,  # 10MB
            resumable=True
        )
        
        request = self.youtube.videos().insert(
            part="snippet,status",
            body={
                "snippet": {
                    "title": metadata['title'],
                    "description": metadata['description'],
                    "tags": metadata['tags']
                },
                "status": {
                    "privacyStatus": metadata['privacy']
                }
            },
            media_body=media
        )
        
        response = None
        while response is None:
            status, response = request.next_chunk()
            if status and progress_callback:
                progress = int(status.progress() * 100)
                progress_callback(progress)
        
        return response['id']

Socket.IO 실시간 진행률

// Frontend
const socket = io();
 
socket.on('upload_progress', (data) => {
    updateProgressBar(data.progress);  // 0-100%
    updateStatus(data.message);
});
 
socket.on('upload_complete', (data) => {
    showSuccess(`업로드 완료: ${data.video_id}`);
    redirectToVideoPage(data.url);
});

특징:

• 대용량 파일(>2GB) 안정적 업로드
• 실시간 진행률 표시 (WebSocket)
• 업로드 실패 시 자동 재시도

2. AI 기반 영상 요약 파이프라인

전체 프로세스

[영상 파일]
    ↓ (FFmpeg)
[음성 추출 (.mp3)]
    ↓ (Whisper API)
[자막 텍스트 (.txt)]
    ↓ (Gemini/GLM)
[구조화 요약 (.md)]
    ↓ (Slack)
[완료 알림]

FFmpeg 음성 추출

# stt_service.py
def extract_audio_from_video(video_path, output_path):
    """
    영상에서 음성만 추출
    """
    command = [
        'ffmpeg',
        '-i', video_path,
        '-vn',  # 비디오 제거
        '-acodec', 'libmp3lame',
        '-ab', '192k',
        '-ar', '44100',
        '-y',  # 덮어쓰기
        output_path
    ]
    
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    
    stdout, stderr = process.communicate()
    
    if process.returncode != 0:
        raise Exception(f'FFmpeg failed: {stderr.decode()}')

Whisper STT API

def send_audio_to_whisper(audio_path):
    """
    음성 파일을 Whisper API로 전송
    """
    with open(audio_path, 'rb') as audio_file:
        files = {'file': audio_file}
        data = {
            'model': 'whisper-large-v3',
            'language': 'ko',
            'response_format': 'text'
        }
        
        response = requests.post(
            WHISPER_API_ENDPOINT,
            files=files,
            data=data,
            timeout=600  # 10분
        )
    
    if response.status_code == 200:
        return response.text
    else:
        raise Exception(f'Whisper failed: {response.text}')

3. Map-Reduce 방식 긴 텍스트 요약

문제: LLM 토큰 제한 (Gemini: 32K, GLM: 128K)

해결: 청크 분할 → 개별 요약 → 재병합

# llm_service.py
def summarize_with_gemini(transcript, video_title):
    """
    긴 텍스트를 Map-Reduce로 요약
    """
    CHUNK_SIZE = 30000  # 30K 글자
    
    # 1단계: Map - 청크별 요약
    chunks = split_text_into_chunks(transcript, CHUNK_SIZE)
    chunk_summaries = []
    
    for i, chunk in enumerate(chunks):
        prompt = f"""
다음은 "{video_title}" 영상의 일부입니다.
핵심 내용을 요약해주세요:
 
{chunk}
"""
        try:
            summary = call_gemini_api(prompt)
            chunk_summaries.append(summary)
        except Exception as e:
            logger.error(f'청크 {i} 요약 실패: {e}')
            # 실패해도 다음 청크 계속 진행
            chunk_summaries.append(f"[요약 실패: {str(e)}]")
    
    # 2단계: Reduce - 통합 요약
    if len(chunk_summaries) == 1:
        return chunk_summaries[0]
    
    combined = "\n\n".join([
        f"## 파트 {i+1}\n{summary}"
        for i, summary in enumerate(chunk_summaries)
    ])
    
    final_prompt = f"""
다음은 "{video_title}" 영상의 파트별 요약입니다.
전체를 종합하여 구조화된 최종 요약을 작성해주세요:
 
{combined}
"""
    
    return call_gemini_api(final_prompt)
 
def call_gemini_api(prompt):
    """
    Gemini API 호출
    """
    response = genai.generate_text(
        model='gemini-pro',
        prompt=prompt,
        temperature=0.7,
        max_output_tokens=2048
    )
    
    return response.text

장점:

• 길이 제한 없이 요약 가능
• 개별 청크 실패 시 나머지 진행
• 단계별 캐싱으로 재시도 효율적

4. YouTube API 할당량 최적화

문제: YouTube API Quota 제한 (일일 10,000 units)

해결: 변경 사항만 감지하여 최소 API 호출

# recent_videos_manager.py
def bulk_update_videos(video_updates):
    """
    변경 사항만 API 호출
    """
    for video in video_updates:
        changes = detect_changes(video)
        
        if not changes:
            continue  # 변경 없으면 스킵
        
        # 변경된 필드만 업데이트
        body = {}
        if 'title' in changes:
            body.setdefault('snippet', {})['title'] = video['title']
        
        if 'description' in changes:
            body.setdefault('snippet', {})['description'] = video['description']
        
        if 'privacy' in changes:
            body.setdefault('status', {})['privacyStatus'] = video['privacy']
        
        # API 호출
        youtube.videos().update(
            part=','.join(body.keys()),
            body={'id': video['id'], **body}
        ).execute()
 
def detect_changes(video):
    """
    이전 상태와 비교하여 변경 사항 감지
    """
    old = get_cached_video_data(video['id'])
    changes = []
    
    if old['title'] != video['title']:
        changes.append('title')
    if old['description'] != video['description']:
        changes.append('description')
    if old['privacy'] != video['privacy']:
        changes.append('privacy')
    
    return changes

절감 효과:

• 불필요한 API 호출 80% 감소
• Quota 초과 위험 최소화

5. 근태 관리 자동화

# daou_office_api_v2.py
class DaouOfficeManager:
    def __init__(self, username, password):
        self.session = requests.Session()
        self.username = username
        self.password = password
    
    def auto_check_in(self):
        """
        자동 출근 체크
        """
        # 1. 로그인
        self._login()
        
        # 2. 현재 상태 확인
        status = self._get_current_status()
        
        if status['checked_in']:
            logger.info('이미 출근 체크됨')
            return
        
        # 3. 출근 처리
        response = self.session.post(
            f'{BASE_URL}/api/attendance/check-in',
            json={'timestamp': datetime.now().isoformat()}
        )
        
        # 4. 로컬 폴더 생성 (증빙용)
        create_attendance_folder(datetime.now())
        
        # 5. Slack 알림
        send_slack_notification('✅ 출근 완료')
    
    def _login(self):
        """
        세션 쿠키 획득
        """
        response = self.session.post(
            f'{BASE_URL}/api/login',
            json={
                'username': self.username,
                'password': self.password
            }
        )
        
        if response.status_code != 200:
            raise Exception('로그인 실패')

자동화 효과:

• 출퇴근 체크 누락 0건
• 증빙 폴더 자동 생성
• Slack으로 확인 알림

6. PM2 프로세스 관리

// ecosystem.config.js
module.exports = {
  apps: [{
    name: 'youtube-automation',
    script: 'app.py',
    interpreter: 'python',
    instances: 1,
    autorestart: true,
    watch: false,
    max_memory_restart: '1G',
    env: {
      FLASK_ENV: 'production',
      PORT: 5003
    },
    error_file: './logs/err.log',
    out_file: './logs/out.log',
    log_date_format: 'YYYY-MM-DD HH:mm:ss Z'
  }]
};

장점:

• 자동 재시작 (크래시 시)
• 로그 관리 자동화
• 메모리 사용량 모니터링

🏗 시스템 아키텍처

전체 흐름

[사용자]
    ↓ (웹 UI / 파일 드롭)
[Flask Server]
    ├─ YouTube 업로드
    │   ↓ (완료 후)
    ├─ AI 파이프라인 트리거
    │   ├─ FFmpeg (음성 추출)
    │   ├─ Whisper (STT)
    │   └─ Gemini/GLM (요약)
    ├─ 근태 자동화
    └─ LMS 데이터 수집
        ↓
[Slack 알림]
[로컬 파일 저장]

디렉토리 구조

youtube_auto_upload/
├── app.py                      # 메인 Flask 앱
├── youtube_uploader.py         # YouTube API
├── stt_service.py              # Whisper STT
├── llm_service.py              # Gemini/GLM
├── daou_office_api_v2.py       # 근태 API
├── slack_notifier.py           # Slack 알림
├── safe_logger.py              # 로깅 유틸
│
├── templates/                  # 웹 UI
│   ├── index.html              # 메인 대시보드
│   └── recent_videos.html      # 비디오 관리
│
├── static/                     # CSS/JS
├── reports/                    # 생성된 보고서
├── summaries/                  # AI 요약본
├── transcripts/                # STT 자막
└── saved_thumbnails/           # 썸네일

📊 주요 기능

자동화 기능

기능	설명	절감 시간
YouTube 업로드	메타데이터 자동 입력	영상당 5분
AI 요약	자막 + 구조화 요약	영상당 30분
근태 체크	자동 출퇴근 처리	일일 5분
LMS 수집	다시보기 자동 정리	주당 1시간

API 엔드포인트

파일 관리:

• GET /api/files - 업로드 가능 파일 목록
• POST /upload_file - 영상 업로드 요청

비디오 관리:

• GET /api/recent_videos - 최근 영상 10개
• POST /api/bulk_update_videos - 메타데이터 일괄 수정
• POST /api/batch_update_thumbnail - 썸네일 일괄 변경

AI 서비스:

• POST /api/generate-summary-from-transcript - 요약 재생성
• GET /api/summaries - 요약 파일 목록

근태:

• POST /api/attendance/check-in - 출근
• POST /api/attendance/check-out - 퇴근

📈 성과

시간 절감

• 일일 절감: 평균 2시간
• 주간 절감: 약 10시간
• 연간 절감: 약 500시간

자동화 효과

• YouTube 업로드: 수동 5분 → 자동 30초
• 영상 요약: 수동 30분 → AI 5분
• 근태 체크: 수동 작업 → 자동 0초

품질 향상

• 근태 누락: 0건 (100% 자동화)
• 영상 요약 정확도: 90%+ (Gemini)
• 업로드 실패율: < 1% (재시도 로직)

🔧 기술적 도전 과제

1. 대용량 파일 업로드

문제: 2GB+ 영상 업로드 시간 및 안정성 해결:

• 청크 업로드 (10MB 단위)
• 재개 가능한 업로드 (resumable)
• Socket.IO 실시간 진행률

2. LLM 토큰 제한

문제: Gemini 32K 제한, 긴 영상은 요약 불가 해결:

• Map-Reduce 패턴
• 30K 글자 단위 청크 분할
• 개별 요약 후 재통합

3. YouTube API Quota

문제: 일일 10,000 units 제한 해결:

• 변경 사항만 API 호출
• 로컬 캐시로 중복 호출 방지
• Batch 작업으로 효율화

4. 인코딩 에러

문제: Windows 환경에서 한글 로그 깨짐 해결:

# safe_logger.py
def safe_print(message):
    """
    인코딩 에러 방지
    """
    try:
        print(message)
    except UnicodeEncodeError:
        # cp949로 인코딩 가능한 문자만 출력
        safe_msg = message.encode('cp949', 
                                  errors='ignore').decode('cp949')
        print(safe_msg)

📝 배운 점

Flask 실시간 웹 애플리케이션

1. Socket.IO 통합

   • 실시간 진행률 표시
   • 백그라운드 작업 상태 업데이트
   • 클라이언트-서버 양방향 통신

2. Threading 관리

   • 업로드/AI 처리를 별도 스레드
   • Flask 메인 스레드 블로킹 방지
   • 에러 발생 시 메인 앱 영향 없음

AI 파이프라인 구축

1. Whisper STT 활용

   • 음성 전처리 (FFmpeg)
   • API 타임아웃 관리
   • 긴 오디오 분할 처리

2. LLM 프롬프트 엔지니어링

   • 구조화된 요약 형식 지정
   • Map-Reduce로 긴 텍스트 처리
   • 다중 LLM (Gemini/GLM) 폴백

외부 API 통합

1. YouTube Data API

   • OAuth2 인증 흐름
   • Quota 최적화 전략
   • 재시도 로직 구현

2. 에러 핸들링

   • 각 단계별 try-except
   • 실패 시 로그 + Slack 알림
   • 부분 실패 허용 (청크 요약)

🔮 향후 계획

단기

• SQLite DB 도입 (파일 시스템 → DB)
• Celery로 비동기 작업 큐 교체
• YouTube API Quota 자동 모니터링

중기

• 쇼츠(Shorts) 자동 제작
  • 하이라이트 구간 AI 식별
  • 자동 클리핑 및 업로드
• 다국어 자막 생성
  • Whisper → 번역 API 연동
  • SRT 파일 자동 업로드

장기

• 대시보드 고도화
  • 업로드 통계 차트
  • 시청자 분석 통합
  • 수익 추이 모니터링
• 모바일 앱 (Flutter)
  • 외부에서 업로드 트리거
  • 요약 결과 푸시 알림

---

프로젝트 기간: 2025.01 - 현재 현재 상태: 개인용 (운영 중) 용도: 개인 업무 자동화 최종 업데이트: 2025-12-21

---

⚠️ 주의사항

개인정보 보호:

• 본 프로젝트는 개인용으로 개발됨
• API Keys, 계정 정보는 환경변수로 관리
• 업무 시스템 정보는 일반화하여 표기
• YouTube 채널 정보는 비공개