Spaces:
Running
on
CPU Upgrade
Running
on
CPU Upgrade
| import os | |
| import re | |
| import csv | |
| import sys | |
| import json | |
| import time | |
| import random | |
| import asyncio | |
| import discord | |
| import logging | |
| import os.path | |
| import secrets | |
| import gspread | |
| import datetime | |
| import tempfile | |
| import requests | |
| import threading | |
| import traceback | |
| import gradio_client | |
| import numpy as np | |
| import pandas as pd | |
| import gradio as gr | |
| import plotly.graph_objects as go | |
| from tabulate import tabulate | |
| from requests import HTTPError | |
| from gradio_client import Client | |
| from discord import Color, Embed | |
| from huggingface_hub import HfApi | |
| from discord.ui import Button, View | |
| from discord.ext import commands, tasks | |
| from datetime import datetime, timedelta | |
| from urllib.parse import urlparse, parse_qs | |
| # starting to migrate to HF datasets and away from google sheets | |
| from huggingface_hub import HfApi | |
| from apscheduler.executors.pool import ThreadPoolExecutor | |
| from apscheduler.executors.asyncio import AsyncIOExecutor | |
| from apscheduler.schedulers.asyncio import AsyncIOScheduler | |
| from gspread_formatting.dataframe import format_with_dataframe | |
| discord.utils.setup_logging(level=logging.DEBUG) # discord.py built-in | |
| logging.getLogger("discord.http").setLevel(logging.DEBUG) # HTTP layer | |
| logging.basicConfig( | |
| level=logging.DEBUG, | |
| format="%(asctime)s %(levelname)s %(name)s: %(message)s", | |
| stream=sys.stdout, | |
| ) | |
| DISCORD_TOKEN = os.environ.get("DISCORD_TOKEN", None) | |
| intents = discord.Intents.all() | |
| bot = commands.Bot(command_prefix='!', intents=intents) | |
| GRADIO_APP_URL = "https://huggingface.co/spaces/discord-community/LevelBot" | |
| """""" | |
| bot_ids = [1136614989411655780, 1166392942387265536, 1158038249835610123, 1130774761031610388, 1155489509518098565, 1155169841276260546, 1152238037355474964, 1154395078735953930] | |
| """""" | |
| hf_token = os.environ.get("HF_TOKEN") | |
| discord_loop = None # global variable | |
| class DMButton(Button): | |
| def __init__(self, label, style): | |
| super().__init__(label=label, style=style) | |
| async def callback(self, interaction: discord.Interaction): | |
| # await interaction.user.send(self.message) # this is for DMs, but users may have DMs disabled | |
| user_id = interaction.user.id | |
| guild = interaction.guild | |
| member = guild.get_member(user_id) | |
| pending_role = guild.get_role(1380908157475360899) # pending verification role, this is to verify that the discord account actually clicked the button at least once | |
| if member and pending_role: | |
| await member.add_roles(pending_role) | |
| print(f"Assigned 'Pending Verification' role to {member.name}") # should log this in #admin-logs properly | |
| unique_link = f"<{GRADIO_APP_URL}?user_id={user_id}>" # don't need token | |
| message = f"Login link generated! To complete the verification process:\n- 1 Visit this link,\n- 2 click the 'π€Sign in with Hugging Face' button\n\n{unique_link}" | |
| await interaction.response.send_message(message, ephemeral=True) | |
| async def on_ready(): | |
| try: | |
| print(f'Logged in as {bot.user.name}') | |
| guild = bot.get_guild(879548962464493619) | |
| await guild.chunk() # get all users into bot cache | |
| channel = bot.get_channel(900125909984624713) | |
| if channel: | |
| try: | |
| message = await channel.fetch_message(1271145797433557023) | |
| if message: | |
| button = DMButton(label="Verify Discord Account", style=discord.ButtonStyle.primary) | |
| view = View(timeout=None) | |
| view.add_item(button) | |
| await message.edit(view=view) | |
| print("message edited") | |
| except discord.NotFound: | |
| print(f"Message with ID 1271145797433557023 not found.") | |
| except discord.HTTPException as e: | |
| print(f"Failed to fetch message with ID 1271145797433557023: {e}") | |
| print("------------------------------------------------------------------------") | |
| except Exception as e: | |
| print(f"on_ready Error: {e}") | |
| DISCORD_TOKEN = os.environ.get("DISCORD_TOKEN", None) | |
| def run_bot(): | |
| global discord_loop | |
| discord_loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(discord_loop) | |
| discord_loop.create_task(bot.start(DISCORD_TOKEN)) | |
| discord_loop.run_forever() | |
| threading.Thread(target=run_bot).start() | |
| def verify_button(profile: gr.OAuthProfile | None, request: gr.Request) -> str: | |
| query_params = parse_qs(urlparse(str(request.url)).query) | |
| user_id = query_params.get('user_id', [None])[0] | |
| if not user_id: | |
| return "# β Missing Discord user ID." | |
| if profile is None: | |
| return "# β You're not logged in with Hugging Face." | |
| async def upgrade_role(): | |
| guild = bot.get_guild(879548962464493619) | |
| member = guild.get_member(int(user_id)) | |
| pending_role = guild.get_role(1380908157475360899) # pending | |
| verified_role = guild.get_role(900063512829755413) # verified | |
| if not member: | |
| print(f"β Could not find Discord user {user_id}") | |
| return | |
| if pending_role in member.roles: | |
| await member.remove_roles(pending_role) | |
| await member.add_roles(verified_role) | |
| print(f"β {member.name} verified and upgraded to verified role.") | |
| else: | |
| print(f"β οΈ {member.name} did not have the pending role. Ignoring.") | |
| global discord_loop | |
| if discord_loop: | |
| asyncio.run_coroutine_threadsafe(upgrade_role(), discord_loop) | |
| return f"# β Verification successful! Welcome, {profile.username} π" | |
| demo = gr.Blocks() | |
| with demo: | |
| try: | |
| TITLE = """<h1 align="center" id="space-title">π€ Hugging Face Discord Verification</h1>""" | |
| gr.HTML(TITLE) | |
| with gr.Tabs(elem_classes="tab-buttons") as tabs: | |
| #------------------------------------------------------------------------------ | |
| with gr.TabItem("β Discord Verification", elem_id="verify-tab", id=2): | |
| login_button = gr.LoginButton() | |
| m1 = gr.Markdown() | |
| demo.load(verify_button, inputs=None, outputs=m1) | |
| def check_login_status(): | |
| try: | |
| return login_button.get_session().get("oauth_info", None) | |
| except AttributeError: | |
| return None | |
| def check_login_wrapper(): | |
| session = check_login_status() | |
| if session is None: | |
| return "Not logged in." | |
| else: | |
| return f"Logged in as {session.get('username', 'Unknown')}" | |
| login_button.click(check_login_wrapper, inputs=None, outputs=m1) | |
| #------------------------------------------------------------------------------ | |
| #with gr.TabItem("π Hub-only leaderboard", elem_id="hub-table", id=2): | |
| except Exception as e: | |
| print(f"gradio demo Error: {e}") | |
| demo.queue().launch() |