Spaces:
Paused
Paused
import json | |
import os | |
import re | |
import asyncio | |
import random | |
import datetime | |
from threading import Thread | |
import discord | |
from discord.ext import tasks | |
from PIL import Image | |
import streamlit as st | |
from huggingface_hub import AsyncInferenceClient, login | |
import time | |
import os | |
launch_time = datetime.datetime.utcnow() | |
ph = st.empty() | |
def syncMessages(): | |
if os.listdir("data") == []: | |
return 0 | |
with ph.container(): | |
delta_uptime = datetime.datetime.utcnow() - launch_time | |
hours, remainder = divmod(int(delta_uptime.total_seconds()), 3600) | |
minutes, seconds = divmod(remainder, 60) | |
days, hours = divmod(hours, 24) | |
st.markdown(f"`[Online Time]: [{days:02d}]d | [{hours:02d}]h | [{minutes:02d}]m | [{seconds:02d}]s`") | |
dirs = st.tabs(os.listdir("data")) | |
i = -1 | |
for dir in os.listdir("data"): | |
i += 1 | |
with dirs[i]: | |
if os.listdir(f"data/{dir}") == []: | |
return 0 | |
files = st.tabs(os.listdir("data/" + dir)) | |
k = -1 | |
for file in os.listdir("data/" + dir): | |
k += 1 | |
with files[k]: | |
with open(f"data/{dir}/{file}", "r") as f: | |
o = f.read().split("<|end_of_turn|>") | |
for item in o: | |
if item == "": | |
continue | |
item = item.split(": ", 1) | |
st.markdown(f":blue[{item[0].split('GPT4 Correct ')[1]}]: {item[1]}") | |
lock_file_path = "test.txt" | |
if not os.path.exists(lock_file_path): | |
with open(lock_file_path, "w") as f: | |
f.write("Running!") | |
clone = False | |
st.markdown("Bot is running, reload the page to see activity.") | |
else: | |
print("Streamlit app is already running, only streaming activity.") | |
clone = True | |
while True: | |
syncMessages() | |
time.sleep(0.5) | |
exit() | |
try: | |
os.mkdir("data") | |
except: | |
pass | |
HF_TOKEN = os.environ["HF_TOKEN"] | |
login(HF_TOKEN) | |
SD = AsyncInferenceClient(model="dataautogpt3/ProteusV0.2") | |
LLM = AsyncInferenceClient(model="openchat/openchat-3.5-0106") | |
RF = AsyncInferenceClient(model="stabilityai/stable-diffusion-xl-refiner-1.0") | |
def ec(x, fd="<|image|>", sd="<|image|>"): | |
matches = re.findall(re.escape(fd) + "(.*?)" + re.escape(sd), x) | |
matches = matches if matches else [""] | |
return matches | |
intents = discord.Intents.default() | |
intents.message_content = True | |
client = discord.Client(intents=intents) | |
async def on_ready(): | |
if clone: | |
print(f"Clone started.") | |
syncMessages() | |
else: | |
print(f"Logged in as {client.user}") | |
presence.start() | |
async def presence(): | |
if not clone: | |
delta_uptime = datetime.datetime.utcnow() - launch_time | |
hours, remainder = divmod(int(delta_uptime.total_seconds()), 3600) | |
minutes, seconds = divmod(remainder, 60) | |
days, hours = divmod(hours, 24) | |
print(f"Online Time: {days:02d}d | {hours:02d}h | {minutes:02d}m | {seconds:02d}s") | |
await client.change_presence( | |
status=discord.Status.idle, | |
activity=discord.Activity( | |
type=discord.ActivityType.playing, | |
large_image="https://i.imgur.com/Kk2BvJg.jpg", | |
large_text="This is Game Icon", | |
name="Escaping the IRS.", | |
details="", | |
state=f"Running for {days:02d}d | {hours:02d}h | {minutes:02d}m", | |
), | |
) | |
async def on_disconnect(): | |
print("Disconnected, Clone status:", clone) | |
async def on_message(message): | |
for user in message.mentions: | |
message.content = message.content.replace(f"<@{user.id}>", | |
f"<@{str(user)}>") | |
try: | |
msgchannel = message.channel | |
try: | |
msgchannel_name = msgchannel.name | |
guild = message.guild | |
guild_name = guild.name | |
except: | |
guild_name = "Direct" | |
msgchannel_name = message.author.display_name | |
s = f":green[{message.author}]: :violet[{message.content}] :blue[{msgchannel_name}] :orange[{guild_name}]" | |
if message.author == client.user: | |
return | |
sysp = """GPT4 Correct System: You are lr, a discord bot. You were made by Araeyn. | |
You must use the markdown format for your responses. | |
Do not excessively use bullet points. | |
Use emojis at the start of your responses. | |
Use <|title|> at the start of your title for the response and <|title|> at the end of the title. | |
Your title must summarize the user's message. | |
If the user asks you to generate an image, use the start tag <|image|> and the end tag <|image|> around the prompt to generate it. Put this at the end of your response. Do not use a link for the image. | |
For example, if the user asks you to generate an image of a cat, you would say <|image|>A cute cat with long fur that is looking out a window with curious eyes.<|image|> | |
Use the keywords 'best quality, HD, ~*~aesthetic~*~' in your prompt to improve it. | |
Always include a title. | |
Do not generate images unless the user speciifes that they want an image. | |
Talk like the user does,""" | |
try: | |
os.mkdir("data/" + guild_name) | |
except: | |
pass | |
if os.path.exists(f"data/{guild_name}/{msgchannel_name}"): | |
with open(f"data/{guild_name}/{msgchannel_name}", "a") as f: | |
n = "\n" | |
f.write( | |
f"""GPT4 Correct {message.author.name}: {message.content.strip(n)}<|end_of_turn|>""" | |
) | |
else: | |
with open(f"data/{guild_name}/{msgchannel_name}", "w") as f: | |
f.write( | |
f"GPT4 Correct system: {sysp}<|end_of_turn|>GPT4 Correct {message.author}: {message.content}<|end_of_turn|>" | |
) | |
with open(f"data/{guild_name}/{msgchannel_name}", "r") as f: | |
context = f.read() | |
if "Feeway" in guild_name: | |
async with msgchannel.typing(): | |
context += f"GPT4 Correct Assistant:" | |
load = random.choice( | |
[ | |
"https://cdn.dribbble.com/users/744913/screenshots/4094897/media/771a495231b798c0ccf7a59a19f31946.gif", | |
"https://cdn.dribbble.com/users/563824/screenshots/3633228/media/b620ccb3ae8c14ea5447d159ebb1da58.gif", | |
"https://cdn.dribbble.com/users/563824/screenshots/4155980/media/d3828cd14ed415eb6f90310991e06f27.gif", | |
"https://cdn.dribbble.com/users/107759/screenshots/3498589/media/5bc45101de34a80ea71238a02f3a75b5.gif", | |
] | |
) | |
output = await LLM.text_generation(context, | |
stop_sequences=["<|end_of_turn|>"], | |
max_new_tokens=4096) | |
title = ec(output, "<|title|>", "<|title|>")[0] | |
imgp = ec(output)[0] | |
with open(f"data/{guild_name}/{msgchannel_name}", "a") as f: | |
f.write(f"GPT4 Correct Assistant: {output}<|end_of_turn|>") | |
embed = discord.Embed(title=title, | |
description=output.replace( | |
f"<|title|>{title}<|title|>", "").replace(f"<|image|>{imgp}<|image|>", ""), | |
color=0x1E81B0) | |
if imgp != "": | |
embed.set_image(url=load) | |
embed.set_footer( | |
text= | |
"""Creating...""" | |
) | |
else: | |
embed.set_footer( | |
text= | |
"""Information or code generated by Lyre may not always be correct. Lyre was made by Araeyn.""" | |
) | |
e = await message.reply(embed=embed) | |
if imgp != "": | |
try: | |
image = await SD.text_to_image(imgp) | |
image.save("image.png") | |
embed.set_footer( | |
text= | |
"""Refining...""" | |
) | |
file = discord.File("image.png", filename="image.png") | |
embed.set_image(url="attachment://image.png") | |
await e.edit(embed=embed, attachments=[file]) | |
image = await RF.image_to_image("image.png") | |
embed.set_footer( | |
text= | |
"""Information or code generated by Lyre may not always be correct. Lyre was made by Araeyn.""" | |
) | |
image.save("image.png") | |
file = discord.File("image.png", filename="image.png") | |
embed.set_image(url="attachment://image.png") | |
await e.edit(embed=embed, attachments=[file]) | |
except: | |
print(image) | |
except Exception as e: | |
print(e) | |
return 0 | |
token = os.environ["TOKEN"] | |
client.run(token) | |