无需Discord Bot及Discord.py实现Discord消息接收机制求助

我想开发一个无需登录Discord账号就能收发消息的程序，目前已经完成消息发送功能和UI界面，但不知道怎么实现消息接收机制。不想用基于Discord.py bot的复杂方案，希望有人指导不依赖Discord.py获取消息的方法，或者清晰解释可行的方案。

已实现的发送功能代码 #

import requests
import tkinter as tk
import tkinter.ttk as ttk
from tkinter import messagebox
import json

#* UI system

window = tk.Tk()
window.title("Python Discord message sender")
window.geometry("400x250+500+250")

group_frame = tk.Frame(window, width=400, height=115)
group_frame.pack()

group_lb = tk.Label(group_frame, text="Choose group: ", font=("Sans-serif", 13))
group_lb.place(x=130, y=20)

choices = list(json.load(open("urls.json")).keys())

groups = ttk.Combobox(group_frame, values=choices, state="readonly", font=("Sans-serif", 13))
groups.place(x=130, y=50)

msg_frame = tk.Frame(window, width=400, height=135)
msg_frame.pack()

msg_lb = tk.Label(msg_frame, text="Type message to this server: ", font=("Sans-serif", 13))
msg_lb.place(x=100, y=0)

msg_entry = tk.Entry(msg_frame, border=3, width=40, font=("Sans-serif", 13))
msg_entry.place(x=10, y=30)

#* Message sending system

urls = json.load(open("urls.json"))

auths = json.load(open("auths.json"))

header = auths['me']

def send():
    group_sending = groups.get()
    if group_sending == "":
        messagebox.showerror("Error!", "You have not chosen a group to send to! Try again")
    message = msg_entry.get()
    if not message == "":
        content = {
            "content": message
        }

        try:
            res = requests.post(urls[group_sending], content, headers=header)
            if res.status_code == 200:
                messagebox.showinfo("Sending info", "Successfully sent message")
                msg_entry.delete(0, "end")
            else:
                messagebox.showerror("ERROR!", f"An error occurred while sending the message, try again. \n \n Error code: {res.status_code}")
        except Exception as bug:
            if str(bug):
                messagebox.showerror("ERROR!", "An error occurred while sending the message, try again. \n \n Error: No connection")
            else:
                messagebox.showerror("ERROR!", f"An error occurred while sending the message, try again. \n \n Error: {bug}")
    else:
        messagebox.showerror("Error!", "You have not written a message to this group!, Try again")
    

msg_submit = tk.Button(msg_frame, text="Send message", font=("Sans-serif", 13), command=send)
msg_submit.place(x=130, y=60)

window.mainloop()

可行的消息接收方案 #

方案1：REST API轮询（简单易实现） #

利用Discord的REST API定期查询频道消息，对比本地记录的最新消息ID，筛选出新消息。

实现步骤

1. 从已有的频道URL中提取频道ID（比如URL为https://discord.com/api/v9/channels/123456/messages，频道ID就是123456）
2. 调用GET /channels/{channel_id}/messages接口，通过after参数只获取比本地最新消息ID更新的内容
3. 在Tkinter中用window.after()定时执行查询逻辑，比如每5秒一次
4. 处理返回的消息数据，更新UI或存储

整合到现有代码的示例

在现有代码中添加以下内容（放在send函数之后，window.mainloop()之前）：

from datetime import datetime

# 存储每个频道的最后消息ID，避免重复获取
last_message_ids = {}

def fetch_new_messages():
    selected_group = groups.get()
    if not selected_group:
        # 未选择频道，延迟后再查询
        window.after(5000, fetch_new_messages)
        return
    
    channel_url = urls[selected_group]
    channel_id = channel_url.split('/')[-2]
    api_url = f"https://discord.com/api/v9/channels/{channel_id}/messages"
    
    params = {}
    if channel_id in last_message_ids:
        params['after'] = last_message_ids[channel_id]
    
    try:
        res = requests.get(api_url, headers=header, params=params)
        if res.status_code == 200:
            messages = res.json()
            if messages:
                # 更新最新消息ID
                last_message_ids[channel_id] = messages[0]['id']
                # 按时间正序处理新消息，这里示例为打印，可改为更新UI
                for msg in reversed(messages):
                    print(f"[{datetime.fromisoformat(msg['timestamp'])}] {msg['author']['username']}: {msg['content']}")
        elif res.status_code == 429:
            # 触发限流，延长查询间隔
            window.after(10000, fetch_new_messages)
            return
    except Exception as e:
        print(f"获取消息失败: {str(e)}")
    
    # 5秒后再次执行查询
    window.after(5000, fetch_new_messages)

# 程序启动1秒后开始轮询
window.after(1000, fetch_new_messages)

注意事项

• 遵守Discord API的请求频率限制，不要设置小于1秒的查询间隔，否则会被限流（返回429状态码）
• 可以根据需求调整limit参数（最大100），控制每次获取的消息数量

方案2：Gateway WebSocket接口（实时接收） #

通过Discord的Gateway WebSocket连接，实时接收频道消息事件，比轮询效率更高，适合需要实时响应的场景。

实现步骤

1. 调用GET /gateway接口获取WebSocket连接地址
2. 使用websockets库建立WebSocket连接，发送包含认证token的身份验证 payload
3. 监听WebSocket消息，处理MESSAGE_CREATE事件（新消息触发）
4. 维护连接心跳，处理重连逻辑，避免连接断开

核心逻辑示例

import asyncio
import websockets
import json
from datetime import datetime

async def listen_to_messages(channel_id, token):
    # 获取Gateway地址
    gateway_res = requests.get("https://discord.com/api/v9/gateway")
    gateway_url = gateway_res.json()['url'] + "?v=9&encoding=json"
    
    async with websockets.connect(gateway_url) as ws:
        # 接收Hello事件，获取心跳间隔
        hello = await ws.recv()
        hello_data = json.loads(hello)
        heartbeat_interval = hello_data['d']['heartbeat_interval'] / 1000
        
        # 发送身份认证
        auth_payload = {
            "op": 2,
            "d": {
                "token": token,
                "intents": 513,  # 包含GUILD_MESSAGES和DIRECT_MESSAGES权限
                "properties": {
                    "$os": "windows",
                    "$browser": "myapp",
                    "$device": "myapp"
                }
            }
        }
        await ws.send(json.dumps(auth_payload))
        
        # 心跳任务
        async def heartbeat():
            while True:
                await asyncio.sleep(heartbeat_interval)
                await ws.send(json.dumps({"op": 1, "d": None}))
        
        asyncio.create_task(heartbeat())
        
        # 监听消息
        while True:
            message = await ws.recv()
            data = json.loads(message)
            if data['op'] == 0 and data['t'] == 'MESSAGE_CREATE':
                msg_data = data['d']
                # 只处理目标频道的消息
                if msg_data['channel_id'] == channel_id:
                    print(f"[{datetime.fromisoformat(msg_data['timestamp'])}] {msg_data['author']['username']}: {msg_data['content']}")

# 在Tkinter中启动异步任务（需整合asyncio与Tkinter事件循环）
token = header['Authorization'].split(' ')[1]
# 替换为你要监听的频道ID
target_channel_id = "1234567890"
asyncio.run(listen_to_messages(target_channel_id, token))

注意事项

• 需要安装websockets库（pip install websockets）
• 必须正确设置intents参数，确保包含消息相关的权限
• 需要处理连接断开后的重连逻辑，以及心跳超时的情况
• Tkinter是单线程GUI框架，需用asyncio与Tkinter事件循环整合，避免阻塞UI

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