ТОП 5 проектов для новичков на Pyrogram

Автор Борис 11 Июн 2023РубрикиPyrogram

Добро пожаловать на мой новый видеоурок на YouTube, где я представляю вам топ 5 проектов для новичков на Pyrogram — мощной библиотеке для разработки Telegram ботов на языке Python.

В этом видео я покажу вам пять интересных проектов, которые вы можете легко создать с помощью Pyrogram и расширить функциональность своего юзер бота.

Автоматическое выделение сообщения: Вы узнаете, как сделать, чтобы ваш бот мог выделять сообщения пользователей жирным, курсивом или подчеркнутым текстом, добавляя больше выразительности и стиля в переписку.
Автоответчик: Я покажу вам, как настроить свой аккаунт на автоматический ответ на сообщения от пользователей. Будете удивлены, насколько полезной может быть эта функция для обработки повторяющихся запросов или в случае вашего временного отсутствия.
Запрет приема голосовых и видеосообщений: Вы узнаете, как настроить аккаунт так, чтобы он блокировал прием голосовых и видеосообщений от пользователей. Кроме того, я покажу вам, как пересылать эти сообщения себе в избранное, чтобы вы все же могли прослушать или просмотреть их позже.
Отлов апдейтов на набор сообщений: Я покажу вам, как отлавливать и обрабатывать апдейты, когда пользователь начинает набирать сообщение, открывая новые возможности для интерактивных функций вашего бота.
Эффект машинного набора текста: Я расскажу вам, как добавить эффект машинного набора текста в ваши сообщения, чтобы создать уникальный и привлекательный стиль общения с пользователями.

Приступайте к созданию своих собственных проектов на Pyrogram и расширяйте возможности вашего user бота и прокачивайте свои навыки.

Исходный код к уроку

import betterlogging as logging
from pyrogram import Client, idle, filters
import asyncio
from pyrogram.types import Message
from pyrogram.enums import ParseMode
from pyrogram.handlers import MessageHandler
api_id = 12345678
api_hash = 'abcdefghijklmnopqrstuvwxyz'


async def outbox_message(client: Client, message: Message):
    await message.edit_text(text=f'<pre>{message.text}</pre>', parse_mode=ParseMode.HTML)


async def start():
    logging.basic_colorized_config(level=logging.INFO)

    client = Client(name='my_account', api_id=api_id, api_hash=api_hash)
    client.add_handler(MessageHandler(outbox_message, filters.me & filters.text))
    await client.start()
    await idle()
    await client.stop()


if __name__ == '__main__':
    asyncio.run(start())

import asyncio
from pyrogram import Client, filters, idle
from pyrogram.enums import ParseMode
from pyrogram.handlers import MessageHandler
from pyrogram.types import Message

api_id = 12345678
api_hash = 'abcdefghijklmnopqrstuvwxyz'


async def get_dice(client: Client, message: Message):
    value_user = message.dice.value
    msg = await client.send_dice(chat_id=message.chat.id, emoji='🎲')
    await asyncio.sleep(5)
    if value_user > msg.dice.value:
        await message.reply(text='Удача на твоей стороне!')
    elif value_user < msg.dice.value:
        await message.reply(text='Ха-ха! Эта партия за мной!')
    else:
        await message.reply(text='Победила дружба. Ещё разок?')


async def start():
    client = Client("my_account", api_id=api_id, api_hash=api_hash)
    client.add_handler(MessageHandler(get_dice, filters.private & filters.dice))
    await client.start()
    await idle()
    await client.stop()


if __name__ == '__main__':
    asyncio.run(start())

import asyncio
from pyrogram import Client, filters, idle
from pyrogram.handlers import MessageHandler
from pyrogram.types import Message

api_id = 12345678
api_hash = 'abcdefghijklmnopqrstuvwxyz'


async def forward_voice_video(client: Client, message: Message):
    await message.forward(chat_id='me')
    await message.delete()
    await message.reply('Данный пользователь запретил отправлять голосовые и видеосообщения :)')


async def start():
    client = Client("my_account", api_id=api_id, api_hash=api_hash)
    client.add_handler(MessageHandler(forward_voice_video, filters.private & (filters.voice | filters.video_note)))
    await client.start()
    await idle()
    await client.stop()


if __name__ == '__main__':
    asyncio.run(start())

import asyncio
from pyrogram import Client, filters, idle, raw, types
from pyrogram.handlers import MessageHandler, RawUpdateHandler
from pyrogram.types import Message
from typing import Dict, Union
api_id = 12345678
api_hash = 'abcdefghijklmnopqrstuvwxyz'


async def user_typing_handler(client: Client, update: raw.base.Update,
                              users: Dict[int, Union[raw.types.User, raw.types.Channel]],
                              chats: Dict[int, Union[types.Chat, raw.types.Channel]]):
    print(update)
    if isinstance(update, raw.types.UpdateUserTyping):
        update: raw.types.UpdateUserTyping
        await client.send_message(chat_id=update.user_id, text='Не надо мне ничего писать, я и так всё знаю :)')


async def start():
    client = Client("my_account", api_id=api_id, api_hash=api_hash)
    client.add_handler(RawUpdateHandler(user_typing_handler))
    await client.start()
    await idle()
    await client.stop()


if __name__ == '__main__':
    asyncio.run(start())

import asyncio
import json
from pprint import pprint
from pyrogram import Client, filters, idle, raw, types
from pyrogram.handlers import MessageHandler, RawUpdateHandler
from pyrogram.types import Message
from typing import Dict, Union
from pyrogram.errors import FloodWait
api_id = 12345678
api_hash = 'abcdefghijklmnopqrstuvwxyz'


data_messages = {}


async def typing_text_handler(client: Client, update: raw.base.Update,
                              users: Dict[int, Union[raw.types.User, raw.types.Channel]],
                              chats: Dict[int, Union[types.Chat, raw.types.Channel]]
                              ):
    if isinstance(update, raw.types.UpdateReadHistoryOutbox):
        update: raw.types.UpdateReadHistoryOutbox
        user_id = data_messages.get(update.peer.user_id)
        if user_id and user_id.get(update.max_id):
            typing_symbol = '|'
            full_text = user_id.get(update.max_id)
            curr_text = full_text[1:]
            text_write = full_text[0]
            while text_write != full_text:
                await edit_text(client=client, chat_id=update.peer.user_id, message_id=update.max_id, text=text_write + typing_symbol)
                await asyncio.sleep(0.02)
                text_write += curr_text[0]
                curr_text = curr_text[1:]
                await edit_text(client=client, chat_id=update.peer.user_id, message_id=update.max_id, text=text_write)
                await asyncio.sleep(0.02)


async def edit_text(client: Client, chat_id: int, message_id: int, text: str):
    try:
        await client.edit_message_text(chat_id=chat_id, message_id=message_id, text=text)
    except FloodWait as e:
        await asyncio.sleep(delay=e.value)
        await edit_text(client=client, chat_id=chat_id, message_id=message_id, text=text)


async def outbox_message(client: Client, message: Message):
    text = message.text.replace('/typing', '').strip()
    await message.edit_text(text=text[0])
    if data_messages.get(message.chat.id):
        data_msg: Dict = data_messages.get(message.chat.id)
        data_msg[message.id] = text
        data_messages[message.chat.id] = data_msg
    else:
        data_messages[message.chat.id] = {message.id: text}

    pprint(data_messages)


async def start():
    client = Client("my_account", api_id=api_id, api_hash=api_hash)
    client.add_handler(MessageHandler(outbox_message, filters.me & filters.text & filters.command('typing')))
    client.add_handler(RawUpdateHandler(typing_text_handler))
    await client.start()
    await idle()
    await client.stop()


if __name__ == '__main__':
    asyncio.run(start())