app.py

import discord
import logging
import os
from huggingface_hub import InferenceClient
import requests
import asyncio
import subprocess

# 로깅 설정
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_API_VIDEO_URL = "https://api.pexels.com/videos/search"

# 특정 채널 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}로 로그인되었습니다!')
        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("찾고싶은 동영상에 대한 설명을 한 문장 단위로 입력하세요. 예) 눈밭에서 뛰어 노는 강아지")

    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)
                if video_urls:
                    # 요청자와의 쓰레드 생성 또는 기존 쓰레드 사용
                    await create_or_use_thread_and_send_videos(message, keywords, video_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": 20  # 최대 20개 동영상을 가져옵니다
    }
    response = requests.get(PEXELS_API_VIDEO_URL, headers=headers, params=params)
    if response.status_code == 200:
        data = response.json()
        video_urls = []
        for video in data['videos']:
            highest_quality_video = max(video['video_files'], key=lambda x: x['width'] * x['height'])
            video_urls.append(highest_quality_video['link'])
        return video_urls
    return None

async def create_or_use_thread_and_send_videos(message, keywords, video_urls):
    # 메시지에 대해 이미 생성된 쓰레드를 찾음
    thread = await find_existing_thread(message)
    
    if not thread:
        # 새 쓰레드 생성
        thread = await message.channel.create_thread(name=f"{message.author.name}의 검색 결과", message=message)
    
    message_content = f"**{keywords}**에 대한 고해상도 동영상 {len(video_urls)}개를 찾았습니다:"
    await thread.send(message_content)
    for url in video_urls:
        await thread.send(url)

async def find_existing_thread(message):
    # 채널의 모든 활성 쓰레드 가져오기
    active_threads = await message.channel.active_threads()
    for thread in active_threads:
        if thread.starter_message_id == message.id:
            return thread
    
    # 채널의 모든 보관된 쓰레드 가져오기
    archived_threads = await message.channel.archived_threads().flatten()
    for thread in archived_threads:
        if thread.starter_message_id == message.id:
            return thread

    return None

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