app.py

import discord
import logging
import os
from huggingface_hub import InferenceClient
import requests
import asyncio
from googleapiclient.discovery import build

# 로깅 설정
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"))

# Pexels API 키 설정
PEXELS_API_KEY = "5woz23MGx1QrSY0WHFb0BRi29JvbXPu97Hg0xnklYgHUI8G0w23FKH62"
PEXELS_VIDEO_API_URL = "https://api.pexels.com/videos/search"

# YouTube API 키 설정
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
youtube_service = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)

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

# 대화 히스토리를 저장할 전역 변수
conversation_history = []

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}로 로그인되었습니다!')

        # 봇이 시작될 때 안내 메시지를 전송
        channel = self.get_channel(SPECIFIC_CHANNEL_ID)
        if channel:
            await channel.send("찾고싶은 비디오나 음악에 대한 설명을 한 문장 단위로 입력하세요. 예) 바다에서 서핑하는 사람 또는 흥겨운 여름 음악")

    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:
            # 의미 분석하여 영문 키워드 추출
            keywords = await extract_keywords(message)
            if keywords:
                # Pexels API로 고해상도 비디오 검색
                video_urls = await search_videos(keywords)
                # YouTube API로 음악 검색
                music_urls = await search_music_on_youtube(keywords)
                if video_urls or music_urls:
                    # 요청자와의 쓰레드 생성 및 결과 전송
                    await create_thread_and_send_results(message, keywords, video_urls, music_urls)
                else:
                    await message.channel.send(f"**{keywords}**에 대한 비디오나 음악을 찾을 수 없습니다.")
            else:
                await message.channel.send("키워드를 추출할 수 없습니다.")
        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
        )

async def extract_keywords(message):
    user_input = message.content
    system_prompt = "다음 문장의 의미에 맞는 영문 키워드를 추출하세요: "
    
    logging.debug(f'Extracting keywords from user input: {user_input}')

    messages = [{"role": "system", "content": system_prompt + user_input}]
    logging.debug(f'Messages to be sent to the model: {messages}')

    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(None, lambda: hf_client.chat_completion(
        messages, max_tokens=10, temperature=0.7, top_p=0.85))

    # Hugging Face 응답 파싱
    if response.choices and response.choices[0].message:
        keywords = response.choices[0].message['content'].strip()
    else:
        keywords = ""
    logging.debug(f'Extracted keywords: {keywords}')
    return keywords

async def search_videos(keywords):
    headers = {
        "Authorization": PEXELS_API_KEY
    }
    params = {
        "query": keywords,
        "per_page": 10  # 최대 10개 비디오를 가져옵니다
    }
    response = requests.get(PEXELS_VIDEO_API_URL, headers=headers, params=params)
    if response.status_code == 200:
        data = response.json()
        return [video['video_files'][0]['link'] for video in data['videos']]
    return None

async def search_music_on_youtube(keywords):
    response = youtube_service.search().list(
        q=keywords,
        part='snippet',
        type='video',
        maxResults=5,
        videoCategoryId='10'  # 음악 카테고리 ID
    ).execute()
    music_urls = []
    for item in response.get('items', []):
        music_urls.append(f"https://www.youtube.com/watch?v={item['id']['videoId']}")
    return music_urls

async def create_thread_and_send_results(message, keywords, video_urls, music_urls):
    # 쓰레드 생성
    thread = await message.channel.create_thread(name=f"{message.author.name}의 검색 결과", message=message)
    if video_urls:
        message_content = f"**{keywords}**에 대한 고해상도 비디오 {len(video_urls)}개를 찾았습니다:"
        await thread.send(message_content)
        for url in video_urls:
            await thread.send(url)
    if music_urls:
        message_content = f"**{keywords}**에 대한 음악 {len(music_urls)}개를 찾았습니다:"
        await thread.send(message_content)
        for url in music_urls:
            await thread.send(url)

if __name__ == "__main__":
    discord_client = MyClient(intents=intents)
    discord_client.run(os.getenv('DISCORD_TOKEN'))