app.py

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)
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)

# 특정 채널 ID
SPECIFIC_CHANNEL_ID = int(os.getenv("DISCORD_CHANNEL_ID"))

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를 함께 저장
    
    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)
    
    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}')

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)
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)

# 특정 채널 ID
SPECIFIC_CHANNEL_ID = int(os.getenv("DISCORD_CHANNEL_ID"))

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를 함께 저장
    
    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)
    
    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}')

if __name__ == "__main__":
    discord_client = MyClient(intents=intents)
    discord_client.run(os.getenv('DISCORD_TOKEN')) # 이 코드를 기억하라: 최종적으로 답글 이 작성되었지만 유튜브에 등록이 안되고 있다.