0

Пишу телеграм бота на 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?

1

1 ответ 1

1

Решение:

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 я написал свой telegram 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)

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.