Update app.py

app.py

@@ -1,18 +1,18 @@
-import os
-import google_auth_oauthlib.flow
-from google.oauth2.credentials import Credentials
-from googleapiclient.discovery import build
-from google.auth.transport.requests import Request
 import discord
 import logging
+import os
 import re
 import asyncio
+import json
 import subprocess
 from huggingface_hub import InferenceClient
+from googleapiclient.discovery import build
+from google.oauth2.credentials import Credentials
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
 from youtube_transcript_api import YouTubeTranscriptApi
 from youtube_transcript_api.formatters import TextFormatter
 from dotenv import load_dotenv
-import time
 
 # 환경 변수 로드
 load_dotenv()
@@ -20,7 +20,6 @@ load_dotenv()
 # JSON_TOKEN.json 파일의 경로
 credentials_path = 'JSON_TOKEN.json'
 token_path = 'token.json'
-auth_code_path = 'auth_code.txt'  # 인증 코드를 저장할 파일 경로
 
 # 로깅 설정
 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s:%(levelname)s:%(name)s:%(message)s', handlers=[logging.StreamHandler()])
@@ -39,50 +38,239 @@ hf_client = InferenceClient("CohereForAI/c4ai-command-r-plus", token=os.getenv("
 SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]
 creds = None
 
-def authorize():
-    flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
-        credentials_path, SCOPES)
-    auth_url, _ = flow.authorization_url(prompt='consent')
-    print('Please go to this URL and finish the authentication: {}'.format(auth_url))
+if os.path.exists(token_path):
+    creds = Credentials.from_authorized_user_file(token_path, SCOPES)
+if not creds or not creds.valid:
+    if creds and creds.expired and creds.refresh_token:
+        creds.refresh(Request())
+    else:
+        flow = InstalledAppFlow.from_client_secrets_file(credentials_path, SCOPES)
+        creds = flow.run_local_server(port=0)
+    with open(token_path, 'w') as token:
+        token.write(creds.to_json())
+
+youtube_service = build('youtube', 'v3', credentials=creds)
 
-    print(f'Enter the authorization code in the file {auth_code_path}')
-    
-    # 파일에 인증 코드가 저장될 때까지 기다림
-    while not os.path.exists(auth_code_path):
-        print("Waiting for authorization code...")
-        time.sleep(5)
+# 특정 채널 ID
+SPECIFIC_CHANNEL_ID = int(os.getenv("DISCORD_CHANNEL_ID"))
 
-    with open(auth_code_path, 'r') as file:
-        code = file.read().strip()
-    
-    flow.fetch_token(code=code)
-    creds = flow.credentials
+class MyClient(discord.Client):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.is_processing = False
+
+    async def on_ready(self):
+        logging.info(f'{self.user}로 로그인되었습니다!')
+
+        # web.py 파일 실행
+        subprocess.Popen(["python", "web.py"])
+        logging.info("Web.py server has been started.")
+
+        # 봇이 시작될 때 안내 메시지를 전송
+        channel = self.get_channel(SPECIFIC_CHANNEL_ID)
+        if channel:
+            await channel.send("유튜브 비디오 URL을 입력하면, 자막과 댓글을 기반으로 답글을 작성합니다.")
+
+    async def on_message(self, message):
+        if message.author == self.user:
+            return
+        if not self.is_message_in_specific_channel(message):
+            return
+        if self.is_processing:
+            return
+        self.is_processing = True
+        try:
+            video_id = extract_video_id(message.content)
+            if video_id:
+                transcript = await get_best_available_transcript(video_id)
+                comments = await get_video_comments(video_id)
+                if comments and transcript:
+                    replies = await generate_replies(comments, transcript)
+                    await create_thread_and_send_replies(message, video_id, comments, replies)
+                    await post_replies_to_youtube(video_id, comments, replies)
+                else:
+                    await message.channel.send("자막이나 댓글을 가져올 수 없습니다.")
+            else:
+                await message.channel.send("유효한 유튜브 비디오 URL을 제공해 주세요.")
+        finally:
+            self.is_processing = False
+
+    def is_message_in_specific_channel(self, message):
+        # 메시지가 지정된 채널이거나, 해당 채널의 쓰레드인 경우 True 반환
+        return message.channel.id == SPECIFIC_CHANNEL_ID or (
+            isinstance(message.channel, discord.Thread) and message.channel.parent_id == SPECIFIC_CHANNEL_ID
+        )
+
+def extract_video_id(url):
+    """
+    YouTube 비디오 URL에서 비디오 ID를 추출합니다.
+    """
+    video_id = None
+    youtube_regex = (
+        r'(https?://)?(www\.)?'
+        '(youtube|youtu|youtube-nocookie)\.(com|be)/'
+        '(watch\?v=|embed/|v/|.+\?v=)?([^&=%\?]{11})')
+
+    match = re.match(youtube_regex, url)
+    if match:
+        video_id = match.group(6)
+    logging.debug(f'Extracted video ID: {video_id}')
+    return video_id
+
+async def get_best_available_transcript(video_id):
+    """
+    YouTube 비디오의 자막을 가져옵니다.
+    """
+    try:
+        # 한국어 자막 시도
+        transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['ko'])
+    except Exception as e:
+        logging.warning(f'Error fetching Korean transcript: {e}')
+        try:
+            # 한국어 자막이 없으면 영어 자막 시도
+            transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['en'])
+        except Exception as e:
+            logging.warning(f'Error fetching English transcript: {e}')
+            try:
+                # 영어 자막도 없으면 다른 언어 자막 시도
+                transcripts = YouTubeTranscriptApi.list_transcripts(video_id)
+                transcript = transcripts.find_manually_created_transcript().fetch()
+            except Exception as e:
+                logging.error(f'Error fetching alternative transcript: {e}')
+                return None
+
+    # 자막 포맷팅
+    formatter = TextFormatter()
+    transcript_text = formatter.format_transcript(transcript)
+    logging.debug(f'Fetched transcript: {transcript_text}')
+    return transcript_text
+
+async def get_video_comments(video_id):
+    """
+    YouTube 비디오의 댓글을 가져옵니다.
+    """
+    comments = []
+    response = youtube_service.commentThreads().list(
+        part='snippet',
+        videoId=video_id,
+        maxResults=100  # 최대 100개의 댓글 가져오기
+    ).execute()
+
+    for item in response.get('items', []):
+        comment = item['snippet']['topLevelComment']['snippet']['textOriginal']
+        comment_id = item['snippet']['topLevelComment']['id']
+        comments.append((comment, comment_id))  # 댓글과 댓글 ID를 함께 저장
     
-    with open(token_path, 'w') as token:
-        token.write(creds.to_json())
+    logging.debug(f'Fetched comments: {comments}')
+    return comments
+
+async def generate_replies(comments, transcript):
+    """
+    댓글과 자막을 기반으로 LLM 답글을 생성합니다.
+    """
+    replies = []
+    for comment, _ in comments:
+        messages = [
+            {"role": "system", "content": f"비디오 자막: {transcript}"},
+            {"role": "user", "content": comment}
+        ]
+        loop = asyncio.get_event_loop()
+        response = await loop.run_in_executor(None, lambda: hf_client.chat_completion(
+            messages, max_tokens=400, temperature=0.7, top_p=0.85))  # max_tokens 값을 조정
+
+        if response.choices and response.choices[0].message:
+            reply = response.choices[0].message['content'].strip()
+        else:
+            reply = "답글을 생성할 수 없습니다."
+        replies.append(reply)
     
-    # 인증 코드 파일 삭제 (보안)
-    os.remove(auth_code_path)
+    logging.debug(f'Generated replies: {replies}')
+    return replies
+
+async def create_thread_and_send_replies(message, video_id, comments, replies):
+    """
+    댓글과 답글을 새로운 쓰레드에 전송합니다.
+    """
+    thread = await message.channel.create_thread(name=f"{message.author.name}의 댓글 답글", message=message)
+    for (comment, _), reply in zip(comments, replies):
+        embed = discord.Embed(description=f"**댓글**: {comment}\n**답글**: {reply}")
+        await thread.send(embed=embed)
+
+async def post_replies_to_youtube(video_id, comments, replies):
+    """
+    생성된 답글을 YouTube 댓글로 게시합니다.
+    """
+    for (comment, comment_id), reply in zip(comments, replies):
+        try:
+            youtube_service.comments().insert(
+                part='snippet',
+                body={
+                    'snippet': {
+                        'parentId': comment_id,
+                        'textOriginal': reply
+                    }
+                }
+            ).execute()
+            logging.debug(f'Posted reply to comment: {comment_id}')
+        except Exception as e:
+            logging.error(f'Error posting reply to comment {comment_id}: {e}')
 
-    return creds
+if __name__ == "__main__":
+    discord_client = MyClient(intents=intents)
+    discord_client.run(os.getenv('DISCORD_TOKEN')) 이 코드를 기억하라: 최종적으로 답글 이 작성되었지만 유튜브에 등록이 안되고 있다.import discord
+import logging
+import os
+import re
+import asyncio
+import json
+import subprocess
+from huggingface_hub import InferenceClient
+from googleapiclient.discovery import build
+from google.oauth2.credentials import Credentials
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
+from youtube_transcript_api import YouTubeTranscriptApi
+from youtube_transcript_api.formatters import TextFormatter
+from dotenv import load_dotenv
+
+# 환경 변수 로드
+load_dotenv()
+
+# JSON_TOKEN.json 파일의 경로
+credentials_path = 'JSON_TOKEN.json'
+token_path = 'token.json'
+
+# 로깅 설정
+logging.basicConfig(level=logging.DEBUG, format='%(asctime)s:%(levelname)s:%(name)s:%(message)s', handlers=[logging.StreamHandler()])
+
+# 인텐트 설정
+intents = discord.Intents.default()
+intents.message_content = True
+intents.messages = True
+intents.guilds = True
+intents.guild_messages = True
+
+# 추론 API 클라이언트 설정
+hf_client = InferenceClient("CohereForAI/c4ai-command-r-plus", token=os.getenv("HF_TOKEN"))
+
+# OAuth 2.0 인증 설정
+SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]
+creds = None
 
-# 기존 토큰을 로드하거나 새로 인증
 if os.path.exists(token_path):
     creds = Credentials.from_authorized_user_file(token_path, SCOPES)
-else:
-    creds = authorize()
-
-# 토큰 갱신 또는 재인증
 if not creds or not creds.valid:
     if creds and creds.expired and creds.refresh_token:
         creds.refresh(Request())
     else:
-        creds = authorize()
+        flow = InstalledAppFlow.from_client_secrets_file(credentials_path, SCOPES)
+        creds = flow.run_local_server(port=0)
+    with open(token_path, 'w') as token:
+        token.write(creds.to_json())
 
-# YouTube API 클라이언트 생성
 youtube_service = build('youtube', 'v3', credentials=creds)
 
-# 디스코드 봇 설정
+# 특정 채널 ID
 SPECIFIC_CHANNEL_ID = int(os.getenv("DISCORD_CHANNEL_ID"))
 
 class MyClient(discord.Client):
@@ -233,7 +421,7 @@ async def post_replies_to_youtube(video_id, comments, replies):
     """
     for (comment, comment_id), reply in zip(comments, replies):
         try:
-            response = youtube_service.comments().insert(
+            youtube_service.comments().insert(
                 part='snippet',
                 body={
                     'snippet': {
@@ -242,11 +430,10 @@ async def post_replies_to_youtube(video_id, comments, replies):
                     }
                 }
             ).execute()
-            logging.debug(f'Posted reply to comment: {comment_id} with response: {response}')
+            logging.debug(f'Posted reply to comment: {comment_id}')
         except Exception as e:
             logging.error(f'Error posting reply to comment {comment_id}: {e}')
-            logging.debug(f'Response: {e.resp} | Content: {e.content}')
 
 if __name__ == "__main__":
     discord_client = MyClient(intents=intents)
-    discord_client.run(os.getenv('DISCORD_TOKEN'))
+    discord_client.run(os.getenv('DISCORD_TOKEN')) # 이 코드를 기억하라: 최종적으로 답글 이 작성되었지만 유튜브에 등록이 안되고 있다.