Raspberry Pi で Discord ボットを実行する

Unsplash の Daniel Tafjord によるカバー写真

私は最近ソフトウェア エンジニアリング ブートキャンプを修了し、LeetCode の簡単な質問に取り組み始めました。質問を解決するための毎日のリマインダーがあれば責任を保つのに役立つと感じました。私は、24 時間スケジュールで実行される (もちろん、信頼できる Raspberry Pi 上で) discord ボットを使用してこれを実装することにしました。これは、次のことを実行します。

• 簡単なリートコードの質問の事前定義されたデータバンクに移動します
• Discord チャンネルに投稿されていない質問を取得します
• leetcode の質問を Discord チャンネルのスレッドとして投稿します (解決策を簡単に追加できます)
• チャンネルへの再投稿を避けるため、質問は投稿済みとしてマークされています

LeetCode にアクセスして 1 日 1 問解決する方が簡単かもしれないとは思いますが、このミニプロジェクトでは ChatGPT の助けを借りて Python と Discord について多くのことを学ぶことができました。スケッチノートも初めての試みなので、ご了承ください（笑）

設定

1. Python仮想環境を使用します
2. 依存関係をインストールします
3. Leetcode 簡単な質問データベースをセットアップします
4. 環境変数を設定します
5.Discordアプリを作成します
6. ボットを実行します!

1.Python仮想環境を使用する

最初に Ubuntu 24.04 でこれをテストしたときに、以下のエラーが発生したため、Python 仮想環境の使用をお勧めします

セットアップは比較的簡単です。次のコマンドを実行するだけで、Python 仮想環境が完成します。

python3 -m venv ~/py_envs
ls ~/py_envs  # to confirm the environment was created
source ~/py_envs/bin/activate

2. 依存関係をインストールする

次の依存関係が必要です:

• AWS CLI

以下を実行して AWS CLI をインストールします:

curl -O 'https://awscli.amazonaws.com/awscli-exe-linux-aarch64.zip'
unzip awscli-exe-linux-aarch64.zip 
sudo ./aws/install
aws --version

次に、aws configure を実行して、必要な認証情報を追加します。 「AWS CLI の設定」ドキュメントを参照してください。

• pip の依存関係

次の pip 依存関係は、pip install -rrequirements.txt を実行することで、要件ファイルとともにインストールできます。

# requirements.txt

discord.py
# must install this version of numpy to prevent conflict with
# pandas, both of which are required by leetscrape
numpy==1.26.4   
leetscrape
python-dotenv

3. leetcode簡単な質問データベースをセットアップする

Leetscrape はこのステップに不可欠でした。詳細については、Leetscrape のドキュメントを参照してください。
私は leetcode の簡単な問題にのみ取り組みたいので (私にとってはかなり難しい問題ですらあります)、次のことを行いました:

• leetscrapeを使用してleetcodeからすべての質問のリストを取得し、リストをCSVに保存します

from leetscrape import GetQuestionsList

ls = GetQuestionsList()
ls.scrape() # Scrape the list of questions
ls.questions.head() # Get the list of questions
ls.to_csv(directory="path/to/csv/file")

• Amazon DynamoDB テーブルを作成し、前のステップで保存した CSV からフィルターされた簡単な質問のリストをそこに入力します。

import csv
import boto3
from botocore.exceptions import BotoCoreError, ClientError

# Initialize the DynamoDB client
dynamodb = boto3.resource('dynamodb')

def filter_and_format_csv_for_dynamodb(input_csv):
    result = []

    with open(input_csv, mode='r') as file:
        csv_reader = csv.DictReader(file)

        for row in csv_reader:
            # Filter based on difficulty and paidOnly fields
            if row['difficulty'] == 'Easy' and row['paidOnly'] == 'False':
                item = {
                    'QID': {'N': str(row['QID'])},  
                    'titleSlug': {'S': row['titleSlug']}, 
                    'topicTags': {'S': row['topicTags']},  
                    'categorySlug': {'S': row['categorySlug']},  
                    'posted': {'BOOL': False}  
                }
                result.append(item)

    return result

def upload_to_dynamodb(items, table_name):
    table = dynamodb.Table(table_name)

    try:
        with table.batch_writer() as batch:
            for item in items:
                batch.put_item(Item={
                    'QID': int(item['QID']['N']),  
                    'titleSlug': item['titleSlug']['S'],
                    'topicTags': item['topicTags']['S'],
                    'categorySlug': item['categorySlug']['S'],
                    'posted': item['posted']['BOOL']
                })
        print(f"Data uploaded successfully to {table_name}")

    except (BotoCoreError, ClientError) as error:
        print(f"Error uploading data to DynamoDB: {error}")

def create_table():
    try:
        table = dynamodb.create_table(
            TableName='leetcode-easy-qs',
            KeySchema=[
                {
                    'AttributeName': 'QID',
                    'KeyType': 'HASH'  # Partition key
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'QID',
                    'AttributeType': 'N'  # Number type
                }
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 5,
                'WriteCapacityUnits': 5
            }
        )

        # Wait until the table exists
        table.meta.client.get_waiter('table_exists').wait(TableName='leetcode-easy-qs')
        print(f"Table {table.table_name} created successfully!")

    except Exception as e:
        print(f"Error creating table: {e}")

# Call function to create the table
create_table()

# Example usage
input_csv = 'getql.pyquestions.csv'  # Your input CSV file
table_name = 'leetcode-easy-qs'      # DynamoDB table name

# Step 1: Filter and format the CSV data
questions = filter_and_format_csv_for_dynamodb(input_csv)

# Step 2: Upload data to DynamoDB
upload_to_dynamodb(questions, table_name)

4. 環境変数を設定する

環境変数を保存するための .env ファイルを作成します

DISCORD_BOT_TOKEN=*****

5.Discordアプリを作成する

Discord 開発者ドキュメントの指示に従って、適切な権限を持つ Discord アプリとボットを作成します。少なくとも次の OAuth 権限でボットを承認してください:

• メッセージを送信
• 公開スレッドを作成する
• スレッドでメッセージを送信する

6. ボットを実行してください!

以下は、python3 discord-leetcode-qs.py コマンドで実行できるボットのコードです。

import os
import discord
import boto3
from leetscrape import GetQuestion
from discord.ext import tasks
from dotenv import load_dotenv
import re
load_dotenv()

# Discord bot token
TOKEN = os.getenv('DISCORD_TOKEN')

# Set the intents for the bot
intents = discord.Intents.default()
intents.message_content = True # Ensure the bot can read messages

# Initialize the bot
bot = discord.Client(intents=intents)
# DynamoDB setup
dynamodb = boto3.client('dynamodb')

TABLE_NAME = 'leetcode-easy-qs'
CHANNEL_ID = 1211111111111111111  # Replace with the actual channel ID

# Function to get the first unposted item from DynamoDB
def get_unposted_item():
    response = dynamodb.scan(
        TableName=TABLE_NAME,
        FilterExpression='posted = :val',
        ExpressionAttributeValues={':val': {'BOOL': False}},
    )
    items = response.get('Items', [])
    if items:
        return items[0]
    return None

# Function to mark the item as posted in DynamoDB
def mark_as_posted(qid):
    dynamodb.update_item(
        TableName=TABLE_NAME,
        Key={'QID': {'N': str(qid)}},
        UpdateExpression='SET posted = :val',
        ExpressionAttributeValues={':val': {'BOOL': True}}
    )

MAX_MESSAGE_LENGTH = 2000
AUTO_ARCHIVE_DURATION = 2880

# Function to split a question into words by spaces or newlines
def split_question(question, max_length):
    parts = []
    while len(question) > max_length:
        split_at = question.rfind(' ', 0, max_length)
        if split_at == -1:
            split_at = question.rfind('\n', 0, max_length)
        if split_at == -1:
            split_at = max_length

        parts.append(question[:split_at].strip())
        # Continue with the remaining text
        question = question[split_at:].strip()

    if question:
        parts.append(question)

    return parts

def clean_question(question):
    first_line, _, remaining_question = message.partition('\n')
    return re.sub(r'\n{3,}', '\n', remaining_question)

def extract_first_line(question):
    lines = question.splitlines()
    return lines[0] if lines else ""

# Task that runs on a schedule
@tasks.loop(minutes=1440) 
async def scheduled_task():
    channel = bot.get_channel(CHANNEL_ID)
    item = get_unposted_item()

    if item:
        title_slug = item['titleSlug']['S']
        qid = item['QID']['N']
        question = "%s" % (GetQuestion(titleSlug=title_slug).scrape())

        first_line = extract_first_line(question)
        cleaned_question = clean_message(question)
        parts = split_message(cleaned_question, MAX_MESSAGE_LENGTH)

        thread = await channel.create_thread(
            name=first_line, 
            type=discord.ChannelType.public_thread
        )

        for part in parts:
            await thread.send(part)

        mark_as_posted(qid)
    else:
        print("No unposted items found.")

@bot.event
async def on_ready():
    print(f'{bot.user} has connected to Discord!')
    scheduled_task.start()

@bot.event
async def on_thread_create(thread):
    await thread.send("\nYour challenge starts here! Good Luck!")

# Run the bot
bot.run(TOKEN)

ボットを実行するには複数のオプションがあります。現時点ではこれを tmux シェルで実行しているだけですが、Docker コンテナーや、AWS、Azure、DigitalOcean、またはその他のクラウド プロバイダーの VPC 上でこれを実行することもできます。

あとは実際に Leetcode の問題を解いてみるだけです...

以上がRaspberry Pi で Discord ボットを実行するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。