Spaces:
Running
Running
# This code is based on the following example: | |
# https://discordpy.readthedocs.io/en/stable/quickstart.html#a-minimal-bot | |
import os | |
import discord | |
from discord import app_commands | |
from discord.ext import commands | |
from threading import Thread | |
import json | |
from horde import HordeAPI | |
intents = discord.Intents.default() | |
intents.message_content = True | |
bot = commands.Bot(command_prefix='>', intents=intents) | |
tree = bot.tree | |
async def hello(interaction: discord.Interaction): | |
await interaction.response.send_message(f"Hello, {interaction.user.mention}!") | |
async def greet(interaction: discord.Interaction, name: str): | |
await interaction.response.send_message(f"Hello, {name}!") | |
async def getKudos(interaction: discord.Interaction): | |
await interaction.response.defer() | |
async with HordeAPI.getUserDetails() as details: | |
if "kudos" not in details: | |
await interaction.followup.send(f'Error: {details["code"]} {details["reason"]}') | |
return | |
await interaction.followup.send(f'The amount of Kudos this user has is {details["kudos"]}') | |
async def generateStatus(interaction: discord.Interaction, id: str): | |
await interaction.response.defer() | |
async with HordeAPI.generateCheck(id) as details: | |
if "kudos" not in details: | |
await interaction.followup.send(f'Check Error: {details["code"]} {details["reason"]}') | |
return | |
if bool(details["is_possible"]) == False: | |
await interaction.followup.send("This generation is impossible.") | |
return | |
if bool(details["faulted"]) == True: | |
await interaction.followup.send("This generation is faulted.") | |
return | |
if bool(details["done"]) == True: | |
async with HordeAPI.generateStatus(id) as generationDetail: | |
if "generations" not in generationDetail: | |
await interaction.followup.send(f'Status Error: {generationDetail["code"]} {generationDetail["reason"]}') | |
for i in range(len(generationDetail["generations"])): | |
await interaction.followup.send(generationDetail["generations"][i]["img"]) | |
return | |
if int(details["processing"]) > 0: | |
total = int(details["finished"]) + int(details["processing"]) + int(details["queue_position"]) + int(details["restarted"]) + int(details["waiting"]) | |
await interaction.followup.send(f'Processing image: {details["processing"]}/{total}') | |
return | |
await interaction.followup.send(f'Position in queue: {details["queue_position"]}, wait time: {details["wait_time"]}s') | |
async def ping(ctx): | |
await ctx.send('pong') | |
async def on_ready(): | |
await tree.sync() | |
print('We have logged in as {0.user}'.format(bot)) | |
async def on_message(message): | |
if message.author == bot.user: | |
return | |
if message.content == 'ping': | |
await message.channel.send('pong') | |
if message.content.startswith('$hello'): | |
await message.channel.send('Hello!') | |
async def sendMessageToChannelHelper(data): | |
channel = await bot.fetch_channel(os.environ.get("CHANNEL_ID")) | |
# 创建一个 embed 对象 | |
mTitle = "Empty Title" | |
if "id" in data: | |
mTitle = data["id"] | |
if "log_tag" in data: | |
mTitle = data["log_tag"] | |
mDescription = "Empty Description" | |
if "model" in data: | |
mDescription = data["model"] | |
if "log_message" in data: | |
mDescription = data["log_message"] | |
mColor = 0x00ff00 | |
if ("log_tag" in data or "log_message" in data) and (data["log_level"] == "assert" or data["log_level"] == "error"): | |
mColor = 0xff0000 | |
embed = discord.Embed(title=mTitle, description=mDescription, color=mColor) | |
# 将 fields 数据加入 embed | |
for field in data: | |
if field == "img": | |
embed.set_image(url=data[field]) | |
else: | |
embed.add_field(name=field, value=data[field], inline=True) | |
# 发送 embed 消息 | |
await channel.send(embed=embed) | |
def sendMessageToChannel(data): | |
bot.loop.create_task(sendMessageToChannelHelper(data)) | |
def run(): | |
try: | |
token = os.environ.get("TOKEN") or "" | |
if token == "": | |
raise Exception("Please add your token to the Secrets pane.") | |
bot.run(token) | |
except discord.HTTPException as e: | |
if e.status == 429: | |
print( | |
"The Discord servers denied the connection for making too many requests" | |
) | |
print( | |
"Get help from https://stackoverflow.com/questions/66724687/in-discord-py-how-to-solve-the-error-for-toomanyrequests" | |
) | |
else: | |
raise e | |
def discord_bot(): | |
print("Running discord_bot") | |
run() | |