Discord机器人后台任务'updater'未处理异常求助

问题概况 #

基于Python 3.8开发的Discord机器人，依赖已安装，启动初期正常，但后台定时任务updater触发时抛出「Unhandled exception in internal background task 'updater'」错误。

核心错误原因 #

结合代码逻辑和Discord.py任务机制，定位出5个关键问题：

• 定时任务未捕获异常，触发崩溃后无法自动恢复
• INSTAGRAM_ACCOUNTS配置为空字符串，请求无效Instagram用户名导致数据解析失败
• 网络请求仅捕获JSON解析异常，未覆盖超时、连接错误、HTTP状态码错误等场景
• 机器人未完全就绪时调用bot.get_channel(CHANNEL_ID)可能返回None，导致后续操作报错
• 访问Instagram帖子描述时直接使用索引，未处理无标题的情况，触发索引越界错误

修复步骤 #

1. 给定时任务添加异常捕获与容错 #

修改updater任务，确保出错后能继续运行，同时校验频道有效性：

@tasks.loop(seconds=UPDATE_INTERVAL)
async def updater():
    try:
        channel = bot.get_channel(CHANNEL_ID)
        if channel:
            await update_from_instagram(channel)
        else:
            print("目标频道未找到，请检查CHANNEL_ID是否正确")
    except Exception as e:
        print(f"定时任务updater执行出错: {str(e)}")

2. 修复空Instagram账号配置 #

将INSTAGRAM_ACCOUNTS改为实际用户名列表，并在逻辑中过滤空值：

# 替换为实际Instagram用户名
INSTAGRAM_ACCOUNTS = ['example_user', 'another_user']

在update_from_instagram中添加空值过滤：

for username in INSTAGRAM_ACCOUNTS:
    if not username.strip():
        print("跳过空用户名")
        continue
    # 原有逻辑

3. 增强网络请求异常处理 #

完善get_user_data，覆盖所有网络请求相关错误：

def get_user_data(username):
    try:
        response = requests.get(
            f"https://www.instagram.com/{username}/?__a=1&__d=dis",
            headers=HEADER,
            timeout=TIMEOUT
        )
        response.raise_for_status()  # 捕获404、500等HTTP错误
        return response.json()["graphql"]["user"]
    except requests.exceptions.RequestException as e:
        print(f"请求Instagram API失败 ({username}): {str(e)}")
        return None
    except (KeyError, ValueError) as e:
        print(f"解析Instagram返回数据失败 ({username}): {str(e)}")
        return None

4. 避免Instagram数据结构索引越界 #

修改post_image中获取帖子描述的逻辑，处理无标题场景：

# 获取帖子描述，兼容无标题情况
caption = ""
if image_data.get("edge_media_to_caption", {}).get("edges"):
    caption = image_data["edge_media_to_caption"]["edges"][0]["node"]["text"]

embed = discord.Embed(
    title=f"Neuer Beitrag von @{username}",
    url=f"https://www.instagram.com/p/{image_data['shortcode']}/",
    description=caption,
    color=COLOR
)

5. 确保机器人就绪后启动任务 #

在on_ready中校验频道存在后再启动任务：

@bot.event
async def on_ready():
    print(f'{bot.user} 已连接到Discord!')
    channel = bot.get_channel(CHANNEL_ID)
    if channel:
        updater.start()
        print("定时任务updater已启动")
    else:
        print("未找到目标频道，定时任务未启动，请检查CHANNEL_ID")

完整修复后的代码 #

import discord
import requests
from discord.ext import commands, tasks
from parse import parse

# 机器人Token
TOKEN = '你的机器人Token' # 替换为实际Token
# Instagram用户名列表
INSTAGRAM_ACCOUNTS = ['example_user'] # 替换为实际用户名
# 目标频道ID
CHANNEL_ID = 1002978952362590258 # 替换为实际频道ID
# 消息历史读取限制
MESSAGE_HISTORY_LIMIT = 100
# 消息嵌入颜色
COLOR = 0x05c0ec
# 定时更新间隔（秒）
UPDATE_INTERVAL = 200
# 请求超时时间（秒）
TIMEOUT = 5
# Instagram请求头
HEADER = {
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
        "Accept-Encoding": "de,en-US;q=0.7,en;q=0.3",
        "Connection": "keep-alive",
        "Cookie": "你的Instagram Cookie", # 替换为实际Cookie
        "Host": "www.instagram.com",
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "cross-site",
        "TE": "trailers",
        "Upgrade-Insecure-Requests": "1",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:104.0) Gecko/20100101 Firefox/104.0"
}

bot = commands.Bot(command_prefix="~~", intents=discord.Intents.all())


@bot.event
async def on_ready():
    print(f'{bot.user} 已连接到Discord!')
    channel = bot.get_channel(CHANNEL_ID)
    if channel:
        updater.start()
        print("定时任务updater已启动")
    else:
        print("未找到目标频道，定时任务未启动，请检查CHANNEL_ID")


@bot.event
async def on_message(message):
    await bot.process_commands(message)


@bot.command()
async def ping(ctx):
    print("Ping received")
    await ctx.channel.send('pong')


@bot.command()
async def update(ctx):
    print("强制更新中")
    channel = bot.get_channel(CHANNEL_ID)
    if channel:
        await update_from_instagram(channel)
    else:
        await ctx.send("目标频道未找到，请检查配置")


@tasks.loop(seconds=UPDATE_INTERVAL)
async def updater():
    try:
        channel = bot.get_channel(CHANNEL_ID)
        if channel:
            await update_from_instagram(channel)
        else:
            print("目标频道未找到，请检查CHANNEL_ID是否正确")
    except Exception as e:
        print(f"定时任务updater执行出错: {str(e)}")


async def update_from_instagram(channel):
    last_published_images_codes = []
    async for message in channel.history(limit=MESSAGE_HISTORY_LIMIT):
        for embed in message.embeds:
            code = parse('https://www.instagram.com/p/{}/', embed.url)
            if code is not None:
                last_published_images_codes.append(str(code[0]))
    images_data = {}
    users_data = {}
    for username in INSTAGRAM_ACCOUNTS:
        if not username.strip():
            print("跳过空用户名")
            continue
        user_data = get_user_data(username)
        if user_data is not None:
            users_data[username] = user_data
            images_data[username] = get_last_image_data(user_data)
        else:
            print(f"{username}: 更新失败")

    for username in images_data:
        if images_data[username]["shortcode"] not in last_published_images_codes:
            await post_image(channel, username, users_data[username], images_data[username])
            print(f"{username}: 发现新内容，已发布到Discord")
        else:
            print(f"{username}: 无新内容需要发布")


# 发布Instagram内容到Discord频道
async def post_image(channel, username, user_data, image_data):
    caption = ""
    if image_data.get("edge_media_to_caption", {}).get("edges"):
        caption = image_data["edge_media_to_caption"]["edges"][0]["node"]["text"]

    embed = discord.Embed(
        title=f"Neuer Beitrag von @{username}",
        url=f"https://www.instagram.com/p/{image_data['shortcode']}/",
        description=caption,
        color=COLOR
    )
    embed.set_author(
        name=user_data["full_name"],
        url=f"https://www.instagram.com/{username}/",
        icon_url=user_data["profile_pic_url"]
    )
    embed.set_image(url=image_data["display_url"])
    embed.set_footer(
        text=f"Gefällt {image_data['edge_liked_by']['count']} Mal\n{image_data['edge_media_to_comment']['count']} Kommentare"
    )
    await channel.send(embed=embed)


# 获取Instagram用户数据
def get_user_data(username):
    try:
        response = requests.get(
            f"https://www.instagram.com/{username}/?__a=1&__d=dis",
            headers=HEADER,
            timeout=TIMEOUT
        )
        response.raise_for_status()
        return response.json()["graphql"]["user"]
    except requests.exceptions.RequestException as e:
        print(f"请求Instagram API失败 ({username}): {str(e)}")
        return None
    except (KeyError, ValueError) as e:
        print(f"解析Instagram返回数据失败 ({username}): {str(e)}")
        return None


# 获取用户最新Instagram帖子数据
def get_last_image_data(user_data):
    return user_data["edge_owner_to_timeline_media"]["edges"][0]["node"]


if __name__ == "__main__":
    bot.run(TOKEN)

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