如何让Flask与Discord.py通信，调用异步函数获取DM附件

方案一：异步HTTP接口（最直接高效） #

让Discord机器人启动一个轻量异步HTTP服务，暴露接口供Flask调用，bot可以在正常运行的同时处理HTTP请求并执行内部异步函数。

实现步骤 #

1. 在bot.py中引入aiohttp，添加HTTP服务逻辑
2. 将get_dm_attachments返回的Attachment对象转成可序列化的字典（HTTP接口无法直接返回复杂对象）
3. Flask通过requests库调用bot的HTTP接口获取结果

bot.py 代码示例

import discord
from discord.ext import commands
import aiohttp
from aiohttp import web

bot = commands.Bot(command_prefix='!', intents=discord.Intents.all())

async def get_dm_attachments(user_id):
    user = await bot.fetch_user(user_id)
    if not user:
        return None
    dm = user.dm_channel
    if dm is None:
        dm = await user.create_dm()

    reslist = []
    async for message in dm.history(limit=10000):
        if message.attachments:
            for attachment in message.attachments:
                reslist.append({
                    'id': attachment.id,
                    'url': attachment.url,
                    'filename': attachment.filename,
                    'size': attachment.size
                })
    return reslist

# 处理HTTP请求的接口函数
async def handle_attachments_request(request):
    user_id = request.rel_url.query.get('user_id')
    if not user_id:
        return web.json_response({'error': '缺少user_id参数'}, status=400)
    try:
        user_id = int(user_id)
        attachments = await get_dm_attachments(user_id)
        return web.json_response({'attachments': attachments})
    except Exception as e:
        return web.json_response({'error': str(e)}, status=500)

# 启动HTTP服务
async def start_http_server():
    app = web.Application()
    app.add_routes([web.get('/get-attachments', handle_attachments_request)])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print('HTTP服务已启动：http://localhost:8080')

# Bot启动时同时启动HTTP服务
@bot.event
async def on_ready():
    print(f'Bot已登录：{bot.user}')
    await start_http_server()

bot.run('你的Discord机器人令牌')

app.py 代码示例

from flask import Flask, jsonify
import requests

app = Flask(__name__)

@app.route('/discord-attachments/<int:user_id>')
def fetch_discord_attachments(user_id):
    try:
        response = requests.get(f'http://localhost:8080/get-attachments?user_id={user_id}')
        response.raise_for_status()
        return jsonify(response.json())
    except requests.exceptions.RequestException as e:
        return jsonify({'error': f'请求失败：{str(e)}'}), 500

if __name__ == '__main__':
    app.run(debug=True, port=5000)

注意事项 #

• 可添加密钥验证防止非法调用接口
• 确保bot和Flask在同一网络环境下可互相访问

方案二：MySQL消息队列（适合非实时场景） #

利用现有MySQL作为中间层，Flask写入请求记录，bot定时轮询数据库处理请求并返回结果。

实现步骤 #

1. 创建MySQL表存储请求状态和结果
2. Flask插入请求记录，轮询等待结果
3. Bot定时查询待处理请求，执行get_dm_attachments后更新数据库状态

第一步：创建MySQL表

CREATE TABLE discord_attachment_requests (
    id INT AUTO_INCREMENT PRIMARY KEY,
    user_id BIGINT NOT NULL,
    status ENUM('pending', 'completed', 'failed') DEFAULT 'pending',
    result TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

bot.py 代码示例

import discord
from discord.ext import commands, tasks
import mysql.connector
import json

bot = commands.Bot(command_prefix='!', intents=discord.Intents.all())

# 数据库配置
DB_CONFIG = {
    'host': 'localhost',
    'user': '你的MySQL用户名',
    'password': '你的MySQL密码',
    'database': '你的数据库名'
}

async def get_dm_attachments(user_id):
    # 同方案一的函数实现，返回字典列表
    user = await bot.fetch_user(user_id)
    if not user:
        return None
    dm = user.dm_channel
    if dm is None:
        dm = await user.create_dm()

    reslist = []
    async for message in dm.history(limit=10000):
        if message.attachments:
            for attachment in message.attachments:
                reslist.append({
                    'id': attachment.id,
                    'url': attachment.url,
                    'filename': attachment.filename,
                    'size': attachment.size
                })
    return reslist

# 定时轮询处理请求
@tasks.loop(seconds=5)
async def process_db_requests():
    conn = mysql.connector.connect(**DB_CONFIG)
    cursor = conn.cursor(dictionary=True)
    
    # 查询待处理请求
    cursor.execute("SELECT id, user_id FROM discord_attachment_requests WHERE status = 'pending'")
    requests = cursor.fetchall()
    
    for req in requests:
        try:
            attachments = await get_dm_attachments(req['user_id'])
            result_json = json.dumps(attachments)
            cursor.execute("UPDATE discord_attachment_requests SET status = 'completed', result = %s WHERE id = %s",
                          (result_json, req['id']))
        except Exception as e:
            cursor.execute("UPDATE discord_attachment_requests SET status = 'failed', result = %s WHERE id = %s",
                          (str(e), req['id']))
        conn.commit()
    
    cursor.close()
    conn.close()

@bot.event
async def on_ready():
    print(f'Bot已登录：{bot.user}')
    process_db_requests.start()

bot.run('你的Discord机器人令牌')

app.py 代码示例

from flask import Flask, jsonify
import mysql.connector
import json
import time

app = Flask(__name__)

DB_CONFIG = {
    'host': 'localhost',
    'user': '你的MySQL用户名',
    'password': '你的MySQL密码',
    'database': '你的数据库名'
}

@app.route('/discord-attachments/<int:user_id>')
def fetch_discord_attachments(user_id):
    # 插入请求记录
    conn = mysql.connector.connect(**DB_CONFIG)
    cursor = conn.cursor()
    cursor.execute("INSERT INTO discord_attachment_requests (user_id) VALUES (%s)", (user_id,))
    req_id = cursor.lastrowid
    conn.commit()
    cursor.close()
    conn.close()

    # 轮询等待结果，超时30秒
    timeout = time.time() + 30
    while time.time() < timeout:
        conn = mysql.connector.connect(**DB_CONFIG)
        cursor = conn.cursor(dictionary=True)
        cursor.execute("SELECT status, result FROM discord_attachment_requests WHERE id = %s", (req_id,))
        res = cursor.fetchone()
        cursor.close()
        conn.close()
        
        if res['status'] == 'completed':
            return jsonify(json.loads(res['result']))
        elif res['status'] == 'failed':
            return jsonify({'error': res['result']}), 500
        time.sleep(1)
    
    return jsonify({'error': '请求超时'}), 408

if __name__ == '__main__':
    app.run(debug=True, port=5000)

方案三：共享Asyncio事件循环（无依赖场景） #

让bot的事件循环在单独线程运行，Flask通过线程安全的方式调用bot的异步函数。

bot.py 代码示例

import discord
from discord.ext import commands
import asyncio
import threading

bot = commands.Bot(command_prefix='!', intents=discord.Intents.all())
bot_loop = None  # 存储bot的事件循环

async def get_dm_attachments(user_id):
    # 同方案一的函数实现，返回字典列表
    user = await bot.fetch_user(user_id)
    if not user:
        return None
    dm = user.dm_channel
    if dm is None:
        dm = await user.create_dm()

    reslist = []
    async for message in dm.history(limit=10000):
        if message.attachments:
            for attachment in message.attachments:
                reslist.append({
                    'id': attachment.id,
                    'url': attachment.url,
                    'filename': attachment.filename,
                    'size': attachment.size
                })
    return reslist

def run_bot():
    global bot_loop
    bot_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(bot_loop)
    bot.run('你的Discord机器人令牌')

# 在后台线程启动bot
threading.Thread(target=run_bot, daemon=True).start()

app.py 代码示例

from flask import Flask, jsonify
import asyncio
from bot import bot_loop, get_dm_attachments

app = Flask(__name__)

def run_async_task(func, *args):
    # 线程安全地在bot的事件循环中执行异步函数
    future = asyncio.run_coroutine_threadsafe(func(*args), bot_loop)
    try:
        return future.result(timeout=30)
    except Exception as e:
        future.cancel()
        raise e

@app.route('/discord-attachments/<int:user_id>')
def fetch_discord_attachments(user_id):
    try:
        attachments = run_async_task(get_dm_attachments, user_id)
        return jsonify({'attachments': attachments})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True, port=5000)

注意事项 #

• 需确保bot线程已启动并初始化好事件循环后，再启动Flask
• 超时时间可根据实际需求调整

内容的提问来源于stack exchange，提问作者MDdiamond