Телеграм бот с функцией постинга списка в telegraph

Телеграм бот с функцией постинга списка в telegraph

Hackerland

Всем привет, начну со скрина работы бота:

Начнём с импортов (Бот будет работать на 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)