#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Smart Shorts Maker - Background Worker Crontab에서 1분마다 실행되어 pending 작업을 처리 """ import os import sys import time import pymysql import traceback from pathlib import Path from datetime import datetime # 환경 변수 설정 (www-data 전용 디렉토리) WORKER_RUNTIME_DIR = '/var/www/html/bicorn/ndegree/runtime' os.environ['MPLCONFIGDIR'] = f'{WORKER_RUNTIME_DIR}/matplotlib' os.environ['XDG_CACHE_HOME'] = f'{WORKER_RUNTIME_DIR}/.cache' os.environ['XDG_RUNTIME_DIR'] = f'{WORKER_RUNTIME_DIR}/xdg' os.environ['PULSE_RUNTIME_PATH'] = f'{WORKER_RUNTIME_DIR}/pulse' # 런타임 디렉토리 생성 Path(WORKER_RUNTIME_DIR).mkdir(parents=True, exist_ok=True) Path(f'{WORKER_RUNTIME_DIR}/matplotlib').mkdir(parents=True, exist_ok=True) Path(f'{WORKER_RUNTIME_DIR}/.cache').mkdir(parents=True, exist_ok=True) Path(f'{WORKER_RUNTIME_DIR}/xdg').mkdir(parents=True, exist_ok=True) Path(f'{WORKER_RUNTIME_DIR}/pulse').mkdir(parents=True, exist_ok=True) # 프로젝트 경로 추가 sys.path.insert(0, '/var/www/html/bicorn/ndegree') from smart_shorts_maker import SmartShortsConverter # 데이터베이스 설정 DB_HOST = 'localhost' DB_USER = 'root' DB_PASS = 'choi4man1!' DB_NAME = 'ndegree' # 로그 디렉토리 LOG_DIR = '/var/www/html/bicorn/ndegree/logs' Path(LOG_DIR).mkdir(parents=True, exist_ok=True) def log(message): """로그 출력""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_message = f"[{timestamp}] {message}" print(log_message) # 로그 파일에도 저장 log_file = Path(LOG_DIR) / f"worker_{datetime.now().strftime('%Y%m%d')}.log" with open(log_file, 'a') as f: f.write(log_message + '\n') def get_db_connection(): """데이터베이스 연결""" try: conn = pymysql.connect( host=DB_HOST, user=DB_USER, password=DB_PASS, database=DB_NAME, charset='utf8mb4', autocommit=False ) return conn except Exception as e: log(f"❌ DB 연결 실패: {str(e)}") return None def update_status(conn, shorts_id, status, progress=None, current_step=None, **kwargs): """상태 업데이트""" try: cursor = conn.cursor() updates = ['status = %s'] values = [status] if progress is not None: updates.append('progress = %s') values.append(progress) if current_step is not None: updates.append('current_step = %s') values.append(current_step) for key, value in kwargs.items(): updates.append(f"{key} = %s") values.append(value) if status == 'processing': updates.append('started_at = NOW()') elif status == 'completed': updates.append('completed_at = NOW()') query = f"UPDATE shorts_videos SET {', '.join(updates)} WHERE id = %s" values.append(shorts_id) cursor.execute(query, values) conn.commit() cursor.close() # 연결 유지 (ping) conn.ping(reconnect=True) except Exception as e: log(f"⚠️ 상태 업데이트 실패: {str(e)}") conn.rollback() def process_job(job): """작업 처리""" shorts_id = job['id'] source_type = job['source_type'] source_path = job['original_path'] or job['source_url'] duration = job['shorts_duration'] subtitles = bool(job['include_subtitles']) language = job['language'] log(f"🎬 작업 시작: ID={shorts_id}, Type={source_type}") conn = get_db_connection() if not conn: return False start_time = time.time() try: # 상태를 processing으로 변경 update_status(conn, shorts_id, 'processing', 0, '처리 시작') # SmartShortsConverter 초기화 converter = SmartShortsConverter( output_dir="/var/www/html/bicorn/ndegree/smart_shorts_output", openai_api_key=os.getenv('OPENAI_API_KEY') ) # 소스 타입에 따라 처리 if source_type == 'file': log(f"📁 파일 처리: {source_path}") # 파일 존재 확인 if not os.path.exists(source_path): raise FileNotFoundError(f"파일을 찾을 수 없습니다: {source_path}") # 10% - 다운로드 완료 (파일은 이미 있음) update_status(conn, shorts_id, 'processing', 10, '파일 로드 완료') # 20% - 음성 인식 시작 update_status(conn, shorts_id, 'processing', 20, 'Whisper 음성 인식 중...') transcription = converter.transcribe_video(source_path, language=language) # 40% - 음성 인식 완료 update_status(conn, shorts_id, 'processing', 40, '음성 인식 완료') # 동영상 길이 확인 from moviepy.editor import VideoFileClip video = VideoFileClip(source_path) video_duration = video.duration video.close() # 50% - 중요 구간 분석 update_status(conn, shorts_id, 'processing', 50, '중요 구간 분석 중...') scored_segments = converter.analyze_important_segments(transcription, video_duration) # 60% - 구간 선택 update_status(conn, shorts_id, 'processing', 60, '최적 구간 선택 중...') selected_segments = converter.select_segments_for_target_duration( scored_segments, duration ) # 70% - 쇼츠 생성 update_status(conn, shorts_id, 'processing', 70, '쇼츠 동영상 생성 중...') shorts_path = converter.create_smart_shorts( source_path, selected_segments, subtitles ) # 85% - AI 요약 update_status(conn, shorts_id, 'processing', 85, 'ChatGPT 요약 생성 중...') summary = converter.generate_summary_with_gpt(transcription) # 95% - 요약 파일 저장 update_status(conn, shorts_id, 'processing', 95, '요약 파일 저장 중...') summary_file = converter.save_summary_file(summary, selected_segments, shorts_path) else: # YouTube log(f"▶️ YouTube 처리: {source_path}") # 5% - 다운로드 시작 update_status(conn, shorts_id, 'processing', 5, 'YouTube 다운로드 중...') # 진행 상황을 단계별로 업데이트하면서 처리 video_path, video_title = converter.download_youtube_video(source_path) update_status(conn, shorts_id, 'processing', 15, '다운로드 완료', original_title=video_title) # 음성 인식 update_status(conn, shorts_id, 'processing', 25, 'Whisper 음성 인식 중...') transcription = converter.transcribe_video(video_path, language=language) update_status(conn, shorts_id, 'processing', 45, '음성 인식 완료') # 동영상 정보 from moviepy.editor import VideoFileClip video = VideoFileClip(video_path) video_duration = video.duration video.close() # 중요 구간 분석 update_status(conn, shorts_id, 'processing', 55, '중요 구간 분석 중...') scored_segments = converter.analyze_important_segments(transcription, video_duration) # 구간 선택 update_status(conn, shorts_id, 'processing', 65, '최적 구간 선택 중...') selected_segments = converter.select_segments_for_target_duration( scored_segments, duration ) # 쇼츠 생성 update_status(conn, shorts_id, 'processing', 75, '쇼츠 동영상 생성 중...') shorts_path = converter.create_smart_shorts( video_path, selected_segments, subtitles ) # AI 요약 update_status(conn, shorts_id, 'processing', 90, 'ChatGPT 요약 생성 중...') summary = converter.generate_summary_with_gpt(transcription) # 요약 파일 저장 update_status(conn, shorts_id, 'processing', 95, '요약 파일 저장 중...') summary_file = converter.save_summary_file(summary, selected_segments, shorts_path) # 임시 파일 정리 converter.cleanup_temp_files() # 처리 완료 processing_time = int(time.time() - start_time) shorts_size = os.path.getsize(shorts_path) if os.path.exists(shorts_path) else 0 shorts_filename = os.path.basename(shorts_path) summary_file_path = str(Path(shorts_path).with_suffix('.txt')) update_status( conn, shorts_id, 'completed', 100, '완료', shorts_filename=shorts_filename, shorts_path=shorts_path, shorts_size=shorts_size, ai_summary=summary, summary_file=summary_file_path if os.path.exists(summary_file_path) else None, processing_time=processing_time ) log(f"✅ 작업 완료: ID={shorts_id}, 시간={processing_time}초") try: if conn and conn.open: conn.close() except: pass return True except Exception as e: # 에러 처리 processing_time = int(time.time() - start_time) error_message = str(e) error_trace = traceback.format_exc() log(f"❌ 작업 실패: ID={shorts_id}") log(f" 에러: {error_message}") log(f" Trace: {error_trace}") update_status( conn, shorts_id, 'failed', 0, '실패', error_message=error_message, processing_time=processing_time ) try: if conn and conn.open: conn.close() except: pass return False def main(): """메인 함수""" log("=" * 60) log("🚀 Worker 시작") conn = get_db_connection() if not conn: log("❌ DB 연결 실패, 종료") return try: cursor = conn.cursor(pymysql.cursors.DictCursor) # pending 상태인 작업 조회 (생성 시간 순) cursor.execute(""" SELECT * FROM shorts_videos WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1 """) job = cursor.fetchone() cursor.close() if not job: log("ℹ️ 대기 중인 작업 없음") conn.close() return log(f"📋 작업 발견: ID={job['id']}") # 작업 처리 success = process_job(job) if success: log("✅ Worker 완료 (성공)") else: log("⚠️ Worker 완료 (실패)") except Exception as e: log(f"❌ Worker 에러: {str(e)}") log(traceback.format_exc()) finally: try: if conn and conn.open: conn.close() except: pass # 이미 닫혔거나 에러 발생 시 무시 log("=" * 60) if __name__ == '__main__': main()