DRF结合Aiogram向所有用户发送随机消息功能失效求助

核心问题拆解 #

• 缺失必要模块导入：代码里用到了datetime、asyncio、random但未导入，直接导致send_random_message_to_all_users函数无法执行，因此看不到print输出。
• Bot调用方式错误：直接使用Bot.send_message类方法发送消息，aiogram要求必须先实例化Bot对象（传入你的Bot Token），再通过实例调用方法。
• 异常处理缩进错误：await asyncio.sleep(e.timeout)未放在RetryAfter的异常块内，会导致逻辑执行混乱。
• 未创建有效Bot实例：没有初始化带Token的Bot对象，这是发送Telegram消息的必备前提。

修复后的完整代码 #

import asyncio
import datetime
import random
from aiogram import Bot
from aiogram.types import ParseMode
from aiogram.utils import exceptions

from servises.advertisement_services import advertisements_service
from servises.subscription_services import subscription_service

# 替换为你的Telegram Bot Token
BOT_TOKEN = "你的BotToken在这里"
bot = Bot(token=BOT_TOKEN, parse_mode=ParseMode.HTML)

async def send_message(user_id, message):
    try:
        await bot.send_message(chat_id=user_id, text=message)
    except exceptions.BotBlocked:
        print(f"目标用户[ID:{user_id}]: 已拉黑Bot")
    except exceptions.ChatNotFound:
        print(f"目标用户[ID:{user_id}]: 用户ID无效")
    except exceptions.RetryAfter as e:
        print(f"目标用户[ID:{user_id}]: 触发Telegram限流，等待{e.timeout}秒")
        await asyncio.sleep(e.timeout)
        return await send_message(user_id, message)
    except exceptions.TelegramAPIError as e:
        print(f"目标用户[ID:{user_id}]: 发送失败，错误信息: {str(e)}")

async def send_message_to_all_users(message):
    users = subscription_service.get_users()
    for user in users:
        user_id = user['account_id']
        await send_message(user_id, message)

async def send_random_message_to_all_users():
    current_time = datetime.datetime.now().strftime('%H:%M')
    print(f"当前时间: {current_time}")
    if current_time == '17:20':
        response = advertisements_service.all_advertisement()
        messages = response.json()
        # 增加非空判断，避免DRF返回空列表时报错
        if messages:
            message = random.choice(messages)
            await send_message_to_all_users(message)
        else:
            print("从DRF未获取到可用消息")

async def scheduler():
    while True:
        await send_random_message_to_all_users()
        await asyncio.sleep(5)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.create_task(scheduler())
        loop.run_forever()
    finally:
        loop.close()

额外优化提示 #

• 增加消息列表非空判断，防止DRF返回空数据时random.choice抛出异常。
• 主函数增加事件循环的关闭逻辑，保证程序退出时资源正常释放。
• 长期来看，建议用APScheduler这类专业定时库替代sleep循环，避免时间漂移问题。

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