主页 / 问题 / 1442052
Accepted
krubsburger
Asked:2022-08-23 22:15:02 +0000 UTC2022-08-23 22:15:02 +0000 UTC

如何将 aiohttp 与 aioschedule 和 aiogram 结合起来

  • 772

我正在写一个电报机器人aiogram，它的功能之一是显示我在market.csgo.com上的账户余额。该函数工作在异步模式下，发送 API 请求并通过await bot.send_message. 现在功能的启动是通过键盘按键来实现的。编码：

...

async def get_balance(session, profiles_dict, message):
    async with session.get(f'https://market.csgo.com/api/v2/get-money?key={profiles_dict[1][1]}') as resp:
        html_1 = await resp.json()

        each_wallet = int(html_1['money'])

        await bot.send_message(message.from_user.id, f'{profiles_dict[1][0]}: {each_wallet} ₽')


@dp.message_handler(content_types=['text'])
async def main(message):
    profiles = users()

    async with aiohttp.ClientSession(trust_env=True) as session:
        tasks = []

        if message.text == 'Balance 💸':
            await bot.send_message(message.from_user.id, 'Information request. Wait..')

            for i in profiles.items():
                task = asyncio.ensure_future(get_balance(session, i, message, stats))
                tasks.append(task)
            await asyncio.gather(*tasks)

        if message.text == 'Check Ban':
            ...

        if message.text == 'List Accounts':
            ...


if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=False)

但是如何让它被调用，让我们说每个小时？我知道有一个异步版本的调度程序aioschedule，并且看过这个例子。但是他们运行一个没有参数的函数，但我有多达 3 个。我试着这样做：

async def scheduler(session, profiles_dict, message):
    aioschedule.every().day.at("12:00").do(get_balance(session, profiles_dict, message))
    while True:
        await aioschedule.run_pending()
        await asyncio.sleep(1)


async def on_startup(session, profiles_dict, message):
    asyncio.create_task(scheduler(session, profiles_dict, message))


if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=False, on_startup=on_startup(session, profiles_dict, message))

在此处输入图像描述

但显然它不是那样工作的。从这里开始，问题是：如何运行一个aiohttp通过调度程序执行异步请求aioschedule并将结果作为消息发送到电报的函数aiogram？

python aiogram

1 个回答

  1. Best Answer

    解决方案：

    import aiogram
import asyncio
import aiohttp
import aioschedule

...

async def get_balance(session, profiles_dict):
    async with session.get(f'https://market.csgo.com/api/v2/get-money?key={profiles_dict[1][1]}') as resp:
        html = await resp.json()

        each_wallet = int(html['money'])

        await bot.send_message(MY_TELEGRAM_ID,
            f'🟢 <a href="{profiles_dict[1][0]}">{profiles_dict[0]}</a> : <i>{each_wallet}</i>',
            disable_web_page_preview=True, parse_mode=types.ParseMode.HTML)

...

@dp.message_handler(content_types=['text'])
async def main(message):
    profiles = users()

    async with aiohttp.ClientSession(trust_env=True) as session:
        tasks = []

        if message.text == 'Balance 💸':
            await bot.send_message(message.from_user.id, 'Information request. Wait..')

            for i in profiles.items():
                task = asyncio.ensure_future(get_balance(session, i))
                tasks.append(task)
            await asyncio.gather(*tasks)

        if message.text == 'On Sale 💰':
            ...

        if message.text == 'Timeout Items ⌛':
            ...


# Client session get_balance function
async def session_get_balance():
    profiles = users()

    async with aiohttp.ClientSession(trust_env=True) as session:
        tasks = []

        for i in profiles.items():
            task = asyncio.ensure_future(get_balance(session, i))
            tasks.append(task)
        await asyncio.gather(*tasks)


# Schedule functions by time
async def scheduler():
    aioschedule.every().hour.do(session_get_balance)
    while True:
        await aioschedule.run_pending()
        await asyncio.sleep(1)


# Function at start
async def on_startup(_):
    asyncio.create_task(scheduler())


# Launch telegram bot
if __name__ == "__main__":
    executor.start_polling(dp, skip_updates=True, on_startup=on_startup)

    因为这是我的个人机器人，message.from_user.id所以我写了我的电报 ID MY_TELEGRAM_ID。一切都很完美。

    await bot.send_message(MY_TELEGRAM_ID,
            f'🟢 <a href="{profiles_dict[1][0]}">{profiles_dict[0]}</a> : <i>{each_wallet}</i>',
            disable_web_page_preview=True, parse_mode=types.ParseMode.HTML)
    • 1

相关问题