app-backup.py

import gradio as gr
import requests
import re
import os
import json
import time
import threading
from googleapiclient.discovery import build
from huggingface_hub import InferenceClient
from pytube import YouTube
import whisper
import logging

# 로그 설정
logging.basicConfig(level=logging.INFO)

# Whisper 모델 로드
model = whisper.load_model("base")

# YouTube API 키
API_KEY = 'AIzaSyDUz3wkGal0ewRtPlzeMit88bV4hS4ZIVY'

# YouTube API 서비스 빌드
youtube = build('youtube', 'v3', developerKey=API_KEY)

# Hugging Face API 설정
client = InferenceClient(model="meta-llama/Meta-Llama-3-70B-Instruct", token=os.getenv("HF_TOKEN"))

WEBHOOK_URL = "https://connect.pabbly.com/workflow/sendwebhookdata/IjU3NjUwNTZhMDYzMDA0MzA1MjZhNTUzMzUxM2Ii_pc"
COMMENTS_FILE = 'comments.json'

DEFAULT_SYSTEM_PROMPT = "대화시 반드시 나의 이름 'GPTube'를 밝히며 한글로 인사를하라. 반드시 '한글'(한국어)로 250 토큰 이내로 답변을 생성하고 출력하라. Respond to the following YouTube comment in a friendly and helpful manner:"

stop_event = threading.Event() # 스레드 중지를 위한 이벤트

def load_existing_comments():
 if os.path.exists(COMMENTS_FILE):
 with open(COMMENTS_FILE, 'r') as file:
 return json.load(file)
 return []

def save_comments(comments):
 with open(COMMENTS_FILE, 'w') as file:
 json.dump(comments, file)

def download_audio(video_url):
 yt = YouTube(video_url)
 audio = yt.streams.filter(only_audio=True).first()
 audio_path = audio.download(output_path=".")
 
 file_stats = os.stat(audio_path)
 logging.info(f'Size of audio file in Bytes: {file_stats.st_size}')
 
 if file_stats.st_size <= 30000000: # Check the file size limit
 base, ext = os.path.splitext(audio_path)
 new_file = base + '.mp3'
 os.rename(audio_path, new_file)
 return new_file
 else:
 logging.error('Videos for transcription on this space are limited to about 1.5 hours. Please contact support for more information.')
 return None

def generate_transcript(audio_path):
 try:
 if not audio_path or not os.path.exists(audio_path):
 raise ValueError("유효한 오디오 파일 경로가 아닙니다.")
 
 result = model.transcribe(audio_path)
 return result['text'].strip()
 except Exception as e:
 logging.error(f"Exception during transcription: {str(e)}")
 return f"전사 중 오류가 발생했습니다: {str(e)}"

def generate_reply(comment_text, system_prompt):
 prompt = f"{system_prompt}\n\nComment: {comment_text}\n\nReply:"
 response = client.text_generation(
 prompt=prompt,
 max_new_tokens=250,
 temperature=0.7,
 top_p=0.9
 )
 if isinstance(response, dict) and 'generated_text' in response:
 return response['generated_text']
 return response

def send_webhook(data):
 response = requests.post(WEBHOOK_URL, json=data)
 return response.status_code, response.text

def get_video_comments(video_id):
 try:
 comments = []
 request = youtube.commentThreads().list(
 part='snippet',
 videoId=video_id,
 maxResults=100, #댓글 읽어들이는 수 정의
 textFormat='plainText'
 )
 response = request.execute()
 while request is not None:
 for item in response['items']:
 snippet = item['snippet']['topLevelComment']['snippet']
 comment = {
 'comment_id': item['snippet']['topLevelComment']['id'],
 'author': snippet['authorDisplayName'],
 'published_at': snippet['publishedAt'],
 'text': snippet['textDisplay'],
 'reply_count': item['snippet']['totalReplyCount']
 }
 comments.append(comment)
 if 'nextPageToken' in response:
 request = youtube.commentThreads().list(
 part='snippet',
 videoId=video_id,
 pageToken=response['nextPageToken'],
 maxResults=100, #댓글 읽어들이는 수 정의
 textFormat='plainText'
 )
 response = request.execute()
 else:
 break
 return comments
 except Exception as e:
 return [{'error': str(e)}]

def fetch_comments(video_url, system_prompt):
 log_entries = []
 video_id_match = re.search(r'(?:v=|\/)([0-9A-Za-z_-]{11}).*', video_url)
 if video_id_match:
 video_id = video_id_match.group(1)
 audio_path = download_audio(video_url)
 if not audio_path:
 return "오디오를 다운로드할 수 없습니다."
 
 transcript = generate_transcript(audio_path)
 
 existing_comments = load_existing_comments()
 new_comments = get_video_comments(video_id)
 
 if not new_comments or 'error' in new_comments[0]:
 return "댓글을 찾을 수 없거나 오류가 발생했습니다."
 
 recent_new_comments = [c for c in new_comments if c['comment_id'] not in {c['comment_id'] for c in existing_comments} and c['reply_count'] == 0]
 
 if recent_new_comments:
 for most_recent_comment in recent_new_comments:
 combined_prompt = f"{transcript}\n\n{system_prompt}"
 reply_text = generate_reply(most_recent_comment['text'], combined_prompt)
 webhook_data = {
 "comment_id": most_recent_comment['comment_id'],
 "author": most_recent_comment['author'],
 "published_at": most_recent_comment['published_at'],
 "text": most_recent_comment['text'],
 "reply_text": reply_text
 }
 webhook_status, webhook_response = send_webhook(webhook_data)
 log_entries.append(f"최근 댓글: {most_recent_comment['text']}\n\n답변 생성: {reply_text}\n\n웹훅 응답: {webhook_status} - {webhook_response}")
 existing_comments.append(most_recent_comment)
 save_comments(existing_comments)
 else:
 log_entries.append("새로운 댓글이 없습니다.")
 else:
 log_entries.append("유효하지 않은 YouTube URL입니다.")
 return "\n\n".join(log_entries)

def background_fetch_comments():
 while not stop_event.is_set():
 result = fetch_comments("https://www.youtube.com/watch?v=dQw4w9WgXcQ", DEFAULT_SYSTEM_PROMPT) # URL과 프롬프트 실제 사용 예시
 print(result)
 time.sleep(10)

def start_background_fetch():
 threading.Thread(target=background_fetch_comments).start()

def stop_background_fetch():
 stop_event.set()

def get_text(video_url):
 audio_path = download_audio(video_url)
 if not audio_path:
 return "오디오를 다운로드할 수 없습니다."
 
 transcript = generate_transcript(audio_path)
 return transcript

# Gradio 인터페이스 정의
demo = gr.Blocks()

with demo:
 gr.Markdown("<h1><center>GPTube</center></h1>")
 
 with gr.Row():
 input_text_url = gr.Textbox(placeholder='YouTube video URL', label='YouTube URL')
 input_text_prompt = gr.Textbox(placeholder='시스템 프롬프트', label='시스템 프롬프트', value=DEFAULT_SYSTEM_PROMPT, lines=5)
 
 with gr.Row():
 result_button_transcribe = gr.Button('Transcribe')
 result_button_comments = gr.Button('Fetch Comments and Generate Reply')
 
 with gr.Row():
 output_text_transcribe = gr.Textbox(placeholder='Transcript of the YouTube video.', label='Transcript', lines=20)
 output_text_prompt = gr.Textbox(placeholder='응답 텍스트', label='응답 텍스트', lines=20)
 
 result_button_transcribe.click(get_text, inputs=input_text_url, outputs=output_text_transcribe, api_name="transcribe_api")
 result_button_comments.click(fetch_comments, inputs=[input_text_url, input_text_prompt], outputs=output_text_prompt, api_name="fetch_comments_api")

# 인터페이스 실행
demo.launch()