Телеграм бот с функцией постинга списка в telegraph
Всем привет, начну со скрина работы бота:
Начнём с импортов (Бот будет работать на python3.7+):
from aiogram import Bot, Dispatcher, executor, types from aiogram.dispatcher import Dispatcher from aiogram.utils import executor from aiogram.types import ReplyKeyboardRemove, ReplyKeyboardMarkup, KeyboardButton, InlineKeyboardMarkup, InlineKeyboardButton from aiogram.types import ParseMode from aiogram.contrib.fsm_storage.memory import MemoryStorage from aiogram.dispatcher import FSMContext from aiogram.dispatcher.filters import Text from aiogram.dispatcher.filters.state import State, StatesGroup from telegraphapi import Telegraph
Далее пойдет токен и айди админа. Айди админа нужно для того что бы другие пользователи не могли добавлять юзеров в список.
Так же мы создаём место в памяти куда будут заноситься данные от админа. По типу Юзернейма и города.
ADMIN_ID = 11111111 TOKEN = "111111111:xxXXxxXXxXXxXXxXXxXxX-XxXXxXXx" storage = MemoryStorage() bot = Bot(token=TOKEN) dp = Dispatcher(bot, storage=storage)
Далее я решил создать функцию публикации списка в телеграфе.
def tgraph(): telegraph = Telegraph() telegraph.createAccount("RandomName here") a = "" with open('users.txt', 'r', encoding='utf-8') as f: for l in f: a = a + str(l) page = telegraph.createPage("Список пользователей.", html_content="<p>" + a + "<p>") return 'http://telegra.ph/{}'.format(page['path'])
Сначала мы создаём объект класса телеграф, создаём аккаунт с рандомным именем. Затем создаём пустую строку в которую будем записывать юзеров для публикации. затем открываем текстовый файл users.txt и проходим по каждой строке - записывая её значение в переменную. Затем создаём пост в телеграфе с заголовком "Список пользователей" и контентом - нашей строкой. Функция возвращает ссылку на созданный пост.
Далее будем создавать 2 клавиатуры. Одна для админа - вторая для обычного юзера.
but1 = 'Добавить пользователя' but2 = 'Вывести список' markup = types.ReplyKeyboardMarkup(resize_keyboard=True, selective=True) markup.add(but1) markup.add(but2) markup2 = types.ReplyKeyboardMarkup(resize_keyboard=True, selective=True) markup2.add(but2)
Первая клавиатура для админа, что бы он мог выбрать функцию "Добавить пользователя", вторая клавиатура для обычного пользователя который запустил бота.
Далее у нас идёт обработка команды /start
@dp.message_handler(commands=['start']) async def process_start_command(message: types.Message): if message['from']['id'] == ADMIN_ID: await message.reply(f"Привет, выбери действие.", reply_markup=markup, reply=False) else: await message.reply(f"Привет, выбери действие.", reply_markup=markup2)
Здесь мы проверяем - является ли юзер который написал админом, если это не админ то мы отправляем ему вторую клавиатуру с помощью которой юзер может получить список пользователей, не более.
Далее идёт обработка команды "Вывести список"
@dp.message_handler(text='Вывести список') async def cmd_start(message: types.Message): link = tgraph() await message.reply('Вот доступные пользователи: ' + link.replace('telegra.ph', 'tgraph.io'), reply=False)
Тут всё в принципе просто, мы создаём ссылку на функцию с помощью переменной, затем делаем замену telegra.ph ссылки на tgraph.io потому что мне сказали что телеграф в блоке в РФ.
Так же было принято решение сделать команду /users для чатов. Ведь у кого то не официальные и старые клиенты, которые не обрабатывают кнопки. (например мобильный клиент telegraph)
@dp.message_handler(commands=['users']) async def process_start_command(message: types.Message): link = tgraph() await message.reply('Вот доступные пользователи: ' + link.replace('telegra.ph', 'tgraph.io'), reply=False)
Далее идёт машина состояний которую я выдрал с прошлого бота который создавал зашифрованные записки.
сlass kod_key(StatesGroup): word = State() crypted = State()
Затем делаем обработку команды "Добавить пользователя"
@dp.message_handler(text='Добавить пользователя') async def cmd_start(message: types.Message): if message['from']['id'] == ADMIN_ID: await kod_key.word.set() await message.reply('Введи username\nПример: @username', reply=False)
Тут мы проверяем айди написавшего, если админ то идём дальше. Сделано для того что бы всякие умники не могли добавить послания в телеграф документ. Затем мы задаём состояние kod_key.
Далее мы переходим к состоянию kod_key:
@dp.message_handler(state=kod_key.word) async def process_name(message: types.Message, state: FSMContext): async with state.proxy() as data: data['word'] = message.text await message.reply('Введи город\nПример: Рим', reply=False) await kod_key.next()
Здесь не нужно делать проверку на админа. т.к. сюда нельзя попасть если ты не админ. Тут вы задаем переменной data["word"] текст сообщения который напишет юзер. Оно сохраняется в памяти. Так же в конце мы переходим на следующие состояние "crypted".
@dp.message_handler(state=kod_key.crypted) async def process_age(message: types.Message, state: FSMContext): await state.update_data(crypted=message.text) async with state.proxy() as data: with open("users.txt", 'a', encoding='utf-8') as f: f.write(data['word'] + " " + data["crypted"] + "\n") await message.reply(f'Пользователь успешно записан в файл.\nUsername: ' + data['word'] + ' \nГород: ' + data['crypted'], reply=False) await state.finish()
Вот здесь мы уже знаем юзернейм и город, записываем их в обычный файл. Так же заканчиваем state с помощью finish.
Ну и код заканчивается:
if __name__ == '__main__': print("starting") executor.start_polling(dp)
А теперь полный листинг программы:
from aiogram import Bot, Dispatcher, executor, types from aiogram.dispatcher import Dispatcher from aiogram.utils import executor from aiogram.types import ReplyKeyboardRemove, ReplyKeyboardMarkup, KeyboardButton, InlineKeyboardMarkup, InlineKeyboardButton from aiogram.types import ParseMode from aiogram.contrib.fsm_storage.memory import MemoryStorage from aiogram.dispatcher import FSMContext from aiogram.dispatcher.filters import Text from aiogram.dispatcher.filters.state import State, StatesGroup from telegraphapi import Telegraph ADMIN_ID = 111111 TOKEN = "1111111:XXxxXXxXxxXxXxXxXXxXxXXxXX" storage = MemoryStorage() bot = Bot(token=TOKEN) dp = Dispatcher(bot, storage=storage) def tgraph(): telegraph = Telegraph() telegraph.createAccount("Your dream") a = "" with open('users.txt', 'r', encoding='utf-8') as f: for l in f: a = a + str(l) page = telegraph.createPage("Список пользователей.", html_content="<p>" + a + "<p>") return 'http://telegra.ph/{}'.format(page['path']) but1 = 'Добавить пользователя' but2 = 'Вывести список' markup = types.ReplyKeyboardMarkup(resize_keyboard=True, selective=True) markup.add(but1) markup.add(but2) markup2 = types.ReplyKeyboardMarkup(resize_keyboard=True, selective=True) markup2.add(but2) @dp.message_handler(commands=['start']) async def process_start_command(message: types.Message): if message['from']['id'] == ADMIN_ID: await message.reply(f"Привет, выбери действие.", reply_markup=markup, reply=False) else: await message.reply(f"Привет, выбери действие.", reply_markup=markup2) @dp.message_handler(text='Вывести список') async def cmd_start(message: types.Message): link = tgraph() await message.reply('Вот доступные пользователи: ' + link.replace('telegra.ph', 'tgraph.io'), reply=False) @dp.message_handler(commands=['users']) async def process_start_command(message: types.Message): link = tgraph() await message.reply('Вот доступные пользователи: ' + link.replace('telegra.ph', 'tgraph.io'), reply=False) class kod_key(StatesGroup): word = State() crypted = State() @dp.message_handler(text='Добавить пользователя') async def cmd_start(message: types.Message): if message['from']['id'] == ADMIN_ID: await kod_key.word.set() await message.reply('Введи username\nПример: @username', reply=False) @dp.message_handler(state=kod_key.word) async def process_name(message: types.Message, state: FSMContext): async with state.proxy() as data: data['word'] = message.text await message.reply('Введи город\nПример: Рим', reply=False) await kod_key.next() @dp.message_handler(state=kod_key.crypted) async def process_age(message: types.Message, state: FSMContext): await state.update_data(crypted=message.text) async with state.proxy() as data: with open("users.txt", 'a', encoding='utf-8') as f: f.write(data['word'] + " " + data["crypted"] + "\n") await message.reply(f'Пользователь успешно записан в файл.\nUsername: ' + data['word'] + ' \nГород: ' + data['crypted'], reply=False) await state.finish() if __name__ == '__main__': print("starting") executor.start_polling(dp)
Прошу меня простить за мой ужасный код, вы всегда можете модифицировать мой код и использовать как хотите, я не несу за это ответственности.
Телеграм канал: https://t.me/CodingCommunity