Discord定时Ping人机器人随机触发异常求助

问题情况 #

开发的Discord娱乐机器人，本应在指定时间@角色/用户并打印日志，但出现以下异常：

• 随机无规律发起Ping，且仅发送Ping消息，无代码中定义的附带文本（如"get ready guys"）
• 终端未按预期打印日志
• 问题已出现两次，机器人托管在Bot-Hosting平台

原代码（含问题） #

import discord
from datetime import time, datetime, timezone, timedelta, today
from discord.ext import tasks, commands
from time import sleep


def run():
    time_role = time(16,7,0, tzinfo=timezone(timedelta(hours=a certain timezone)))
    time_person = time(16,7,0, tzinfo=timezone(timedelta(hours=another certain timezone)))
    channel_id = # channel id
    server_id = # server id
    #gives the bot message content intent
    intents = discord.Intents.default()
    intents.message_content = True

    TOKEN = "my token obviously"
    client = discord.Client(intents = intents)

                 
    #pings @role A at time_role
    @tasks.loop(time=time_role)
    async def send_time():
        await client.get_channel(channel_id).send("<@&role A id>")
        await client.get_channel(channel_id).send("get ready guys")
        print("sent role A ping at " + today().strftime('%m-%d %H:%M:%S')) 
    #the above two lines were NOT SENT along with the unexpected pings
    
    #pings @person A at time_person
    @tasks.loop(time=time_person)
    async def send_person(): 
        await client.get_channel(channel_id).send("<@person A id>")
        await client.get_channel(channel_id).send("get ready person A")
        print("sent person A ping at " + today().strftime('%m-%d %H:%M:%S'))
    #the above two lines were NOT SENT along with the unexpected pings


    @client.event
    async def on_ready():
        print(f"{client.user} is now online!
time: "today().strftime('%m-%d %H:%M:%S'))
        if not send_time.is_running():
            send_time.start()
            print("started role ping")
        if not send_person.is_running():
            send_person.start()
            print("started person ping")

    client.run(TOKEN)
run()

问题分析与修复方案 #

1. 语法错误导致日志打印失败 #

on_ready事件中的print语句存在语法错误，混用f-string和普通字符串拼接，直接导致日志无法打印，甚至可能引发代码报错：

# 错误代码
print(f"{client.user} is now online!
time: "today().strftime('%m-%d %H:%M:%S'))

# 修复后
@client.event
async def on_ready():
    current_time = datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')
    print(f"{client.user} is now online!\ntime: {current_time}")
    if not send_time.is_running():
        send_time.start()
        print(f"started role ping at {current_time}")
    if not send_person.is_running():
        send_person.start()
        print(f"started person ping at {current_time}")

2. 任务无异常捕获，崩溃后可能引发异常行为 #

原任务未做异常处理，若发送消息失败（如频道不存在、权限不足），任务可能崩溃，导致机器人状态异常。添加异常捕获：

@tasks.loop(time=time_role)
async def send_time():
    try:
        channel = client.get_channel(channel_id)
        if channel:
            await channel.send("<@&role A id>")
            await channel.send("get ready guys")
            current_time = datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')
            print(f"sent role A ping at {current_time}")
        else:
            print(f"Channel {channel_id} not found at {datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')}")
    except Exception as e:
        print(f"Error in send_time task: {e} at {datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')}")

@tasks.loop(time=time_person)
async def send_person():
    try:
        channel = client.get_channel(channel_id)
        if channel:
            await channel.send("<@person A id>")
            await channel.send("get ready person A")
            current_time = datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')
            print(f"sent person A ping at {current_time}")
        else:
            print(f"Channel {channel_id} not found at {datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')}")
    except Exception as e:
        print(f"Error in send_person task: {e} at {datetime.now(timezone.utc).astimezone().strftime('%m-%d %H:%M:%S')}")

3. 排查异常Ping的来源 #

异常Ping仅发送了Ping消息，无附带文本，需排查：

• 机器人Token是否泄露：立即检查Token权限，若怀疑泄露则重新生成Token
• 托管平台是否存在多实例运行：联系平台客服确认是否有进程意外重启或重复启动的情况
• 是否存在未发现的代码逻辑：检查是否有其他任务、命令或事件处理函数触发了Ping

4. 优化任务启动逻辑 #

虽然on_ready中已判断任务是否运行，但可添加全局标记避免重复初始化：

def run():
    # ... 其他代码 ...
    tasks_initialized = False

    @client.event
    async def on_ready():
        nonlocal tasks_initialized
        if tasks_initialized:
            return
        tasks_initialized = True
        # ... 打印日志、启动任务代码 ...

5. 添加状态检查命令 #

方便查看任务运行状态，确认是否有重复任务：

# 将client改为commands.Bot以支持命令
client = commands.Bot(command_prefix="!", intents=intents)

# 添加状态命令
@client.command()
async def task_status(ctx):
    await ctx.send(f"send_time task running: {send_time.is_running()}")
    await ctx.send(f"send_person task running: {send_person.is_running()}")

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