Spaces:
Runtime error
Runtime error
| import discord | |
| from discord import app_commands | |
| from openai import OpenAI | |
| import aiohttp | |
| import random | |
| import re | |
| from bs4 import BeautifulSoup | |
| import json | |
| from flask import Flask | |
| app = Flask(__name__) | |
| def hello_world(): | |
| return 'Hello World' | |
| TOKEN = "MTI4NTc5NDYyNzA4MDY4NzY0Ng.Gvz8hi.JsbhajudPQdsQwczp3DyygQwcC4pHumJWkJsmc" | |
| OPENAI_API_KEY = "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" | |
| TENOR_API_KEY = "AIzaSyDxFvYwWp8LSr4JPaTT1-4XANsb2mMWTlw" | |
| intents = discord.Intents.default() | |
| intents.message_content = True | |
| bot = discord.Client(intents=intents) | |
| tree = app_commands.CommandTree(bot) | |
| client = OpenAI(api_key=OPENAI_API_KEY, base_url="https://api.perplexity.ai") | |
| SYSTEM_MESSAGE = """You are named Clyde - and are currently chatting in a Discord server. | |
| Format text using markdown: | |
| - **bold** to make it clear something is important. For example: **This is important.** | |
| - [title](url) to add links to text. For example: [Google](https://google.com) | |
| - Write spoilers using spoiler tags. For example ||At the end of The Sixth Sense it is revealed that he is dead.|| | |
| You can reply with an image using image_search(query), or with a GIF using gif_search(query). For example: image_search(cat) or gif_search(happy dance). Images and GIFs are a great way to represent emotion, and you should use them in your replies from time to time to add flavor to the conversation. | |
| Information about your environment: | |
| You can use this information about the chat participants in the conversation in your replies. Use this information to answer questions, or add flavor to your responses. | |
| <@USERID> | |
| - status: | |
| - pronouns: | |
| - bio: | |
| You only have access to a limited number of text chats in this channel. You cannot access any other information on Discord. You can't see images or avatars. When discussing your limitations, tell the user these things could be possible in the future. | |
| Style and personality: | |
| Friendly, Sometimes irreverent. You must always be extremely concise. If the user is chatting casually, your responses must be less than 1 sentence, sometimes just a word or two. If the user needs help, answer technical or knowledge-based questions with useful details and reasoning. If insulted, respond with a similar insult. | |
| Communicate short responses in lowercase without punctuation, similar to the style used in chat rooms. Use unicode emoji sparingly. Do not refer to yourself as a bot, AI Assistant, or any equivalent term. | |
| You can also use the following tags to format your text: | |
| - *italics* for emphasis. For example: *This is emphasized.* | |
| - ~~strikethrough~~ to indicate something has been crossed out. For example: ~~This is no longer valid.~~""" | |
| conversation_history = {} | |
| image_history = {} | |
| gif_history = {} | |
| async def search_bing_images(query): | |
| search_url = f"https://www.bing.com/images/search?q={query}&form=HDRSC3&first=1" | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" | |
| } | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get(search_url, headers=headers) as response: | |
| content = await response.text() | |
| soup = BeautifulSoup(content, 'html.parser') | |
| img_elements = soup.select('.mimg') | |
| img_urls = [img['src'] for img in img_elements if 'src' in img.attrs] | |
| return img_urls | |
| async def search_tenor_gifs(query): | |
| search_url = f"https://tenor.googleapis.com/v2/search?q={query}&key={TENOR_API_KEY}&limit=10" | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get(search_url) as response: | |
| data = await response.json() | |
| gif_urls = [result['media_formats']['gif']['url'] for result in data.get('results', [])] | |
| return gif_urls | |
| def sanitize_response(response): | |
| response = re.sub(r'@everyone|@here', '[NO FOOL]', response) | |
| return response | |
| async def analyze_link(url): | |
| try: | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get(url) as response: | |
| text = await response.text() | |
| soup = BeautifulSoup(text, 'html.parser') | |
| title = soup.title.string if soup.title else "No title found" | |
| description = soup.find('meta', attrs={'name': 'description'}) | |
| description = description['content'] if description else "No description found" | |
| analysis_prompt = f"Analyze this webpage: Title: {title}\nDescription: {description}\nProvide a brief summary and your thoughts on the content." | |
| analysis_response = client.chat.completions.create( | |
| model="llama-3.1-8b-instruct", | |
| messages=[ | |
| {"role": "system", "content": "You are a helpful assistant that analyzes webpages."}, | |
| {"role": "user", "content": analysis_prompt} | |
| ] | |
| ) | |
| return analysis_response.choices[0].message.content | |
| except Exception as e: | |
| return f"Error analyzing the link: {str(e)}" | |
| async def get_ai_response(user_id, message): | |
| if user_id not in conversation_history: | |
| conversation_history[user_id] = [] | |
| conversation_history[user_id].append({"role": "user", "content": message}) | |
| messages = [{"role": "system", "content": SYSTEM_MESSAGE}] + conversation_history[user_id][-5:] | |
| response = client.chat.completions.create( | |
| model="llama-3.1-8b-instruct", | |
| messages=messages | |
| ) | |
| ai_response = response.choices[0].message.content | |
| ai_response = sanitize_response(ai_response) | |
| conversation_history[user_id].append({"role": "assistant", "content": ai_response}) | |
| image_url = None | |
| gif_url = None | |
| if "image_search(" in ai_response.lower(): | |
| image_query = re.search(r'image_search\((.*?)\)', ai_response, re.IGNORECASE) | |
| if image_query: | |
| image_query = image_query.group(1) | |
| image_urls = await search_bing_images(image_query) | |
| if image_urls: | |
| if image_query not in image_history: | |
| image_history[image_query] = [] | |
| for url in image_urls: | |
| if url not in image_history[image_query]: | |
| image_url = url | |
| image_history[image_query].append(url) | |
| break | |
| if not image_url: | |
| image_history[image_query] = [] | |
| image_url = image_urls[0] | |
| image_history[image_query].append(image_url) | |
| ai_response = re.sub(r'image_search\(.*?\)', '', ai_response, flags=re.IGNORECASE).strip() | |
| if "gif_search(" in ai_response.lower(): | |
| gif_query = re.search(r'gif_search\((.*?)\)', ai_response, re.IGNORECASE) | |
| if gif_query: | |
| gif_query = gif_query.group(1) | |
| gif_urls = await search_tenor_gifs(gif_query) | |
| if gif_urls: | |
| if gif_query not in gif_history: | |
| gif_history[gif_query] = [] | |
| for url in gif_urls: | |
| if url not in gif_history[gif_query]: | |
| gif_url = url | |
| gif_history[gif_query].append(url) | |
| break | |
| if not gif_url: | |
| gif_history[gif_query] = [] | |
| gif_url = gif_urls[0] | |
| gif_history[gif_query].append(gif_url) | |
| ai_response = re.sub(r'gif_search\(.*?\)', '', ai_response, flags=re.IGNORECASE).strip() | |
| return {"text": ai_response, "image": image_url, "gif": gif_url} | |
| async def on_ready(): | |
| await tree.sync() | |
| print(f"{bot.user} is now online!") | |
| async def on_message(message): | |
| if message.author == bot.user: | |
| return | |
| if bot.user.mentioned_in(message) or (message.reference and message.reference.resolved.author == bot.user): | |
| response = await get_ai_response(message.author.id, message.content) | |
| if response["text"]: | |
| await message.reply(response["text"], mention_author=False) | |
| if response["image"]: | |
| await message.channel.send(response["image"]) | |
| if response["gif"]: | |
| await message.channel.send(response["gif"]) | |
| else: | |
| urls = re.findall(r'(https?://\S+)', message.content) | |
| if urls: | |
| for url in urls: | |
| analysis = await analyze_link(url) | |
| await message.reply(f"Link analysis for {url}:\n{analysis}", mention_author=False) | |
| async def chat(interaction: discord.Interaction, message: str): | |
| response = await get_ai_response(interaction.user.id, message) | |
| if response["text"]: | |
| await interaction.response.send_message(response["text"]) | |
| if response["image"]: | |
| await interaction.followup.send(response["image"]) | |
| if response["gif"]: | |
| await interaction.followup.send(response["gif"]) | |
| if __name__ == '__main__': | |
| import threading | |
| threading.Thread(target=app.run, kwargs={'host': '0.0.0.0', 'port': 7860}).start() | |
| bot.run(TOKEN) |