Телеграм бот для хранения ссылок + БД

Сегодня мы будем делать простого телеграм бота в котором будем хранить ссылки или другую полезную вам информацию.

Вы можете делать бота на любой библиотеки, я же буду делать на aiogram

Если вы тоже хотите делать на aiogram то вам необходим python 3.7 или выше.

Начнём с импортов.

import sqlite3
import re
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor

sqlite3 Стандартная библиотека которая поставляется вместе с питоном, re это регулярные выражения. Они тоже входят в стандарт. Вам нужно установить лишь aiogram. У меня стоит aiogram==2.0.1 .

Начнем с создания базы данных и функции для записи данных в бд.

def bd():
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	cur.execute('CREATE TABLE IF NOT EXISTS links(Link TEXT)')




async def link_writer(link):
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	mass = []
	mass.append(link)
	cur.execute('INSERT INTO links VALUES(?)', mass)
	con.commit()
	cur.close()

Если вас пугает Async в начале функции для записи - не бойтесь, он нужен для корректной записи ботом. Это вы увидите дальше. А пока можем разобрать функции. Первая функция bd() коннектится к базе данных link.bd если её нет то он автоматически её создаст. Далее мы создаем курсор для работы с базой данных. Затем мы делаем запрос cur.execute('CREATE TABLE IF NOT EXISTS links(Link TEXT)')

Который обозначает "Создать таблицу Links если такой нет. И далее названия столбца, в нашем случае это Link с типом данных TEXT.

Для удобного просмотра базы данных могу порекомендовать софт Database .net скачать который можно вот тут : https://fishcodelib.com/database.htm . Софт бесплатный на 100 дней, официальная версия. Подойдет вам для просмотра вашей базы данных.

Теперь разберем вторую функцию для записи строки в базу. Что бы вы каждый раз не мотали вверх я её продублирую ниже

async def link_writer(link):
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	mass = []
	mass.append(link)
	cur.execute('INSERT INTO links VALUES(?)', mass)
	con.commit()
	cur.close()

Здесь мы создаем коннект с базой, создаем курсор, далее создаем пустой массив, ниже добавляем в наш массив то что передали вместе с функцией и после этого записываем с помощью запроса в нашу базу данных, в таблицу Links. Затем нам необходимо записать данные с помощью con.commit() и закрыть соединения с базой данных с помощью cur.close()

Далее нам нужна будет функция которая будет возвращать нам строку с нашими данными из базы данных что бы бот мог нам её отправить.

async def send_base():
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	query = 'SELECT * FROM links'
	cur.execute(query)
	data = cur.fetchall()
	mm = []
	for i in data:
		mm.append(i)
	ww = len(data)
	g = []
	for i in range(ww):
		a = re.sub('|\(|\'|\,|\)', '', str(mm[i]))
		g.append(a)
	c = []
	for i in g:
		q = i + "\n"
		c.append(q)
	val = '\n'.join(c)
	return val

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

Надо было это фиксить так что мы удаляем ненужные нам символы из каждого элемента массива, затем добавляем исправленные элементы в пустой массив, затем с помощью еще одного цикла добавляем к каждому элементу массива "\n" для перехода на новую строку а затем превращаем массив в строку и возвращаем его.

Теперь займемся самым простым, это подключением бота.

TOKEN = "XXXXXXXXXXXXXXXXXXXXXX_XXXXXXXXXXZ"

bot = Bot(token=TOKEN)
dp = Dispatcher(bot)


@dp.message_handler(commands=['ad'])
async def process_start_command(message: types.Message):
	if message.from_user.id == 1111111111111:
		myString = ' '.join(message.text.split(" ")[1:])
		await link_writer(myString)
		await bot.send_message(message.chat.id, "Запись успешно добавлена в базу.")
	else:
		await bot.send_message(message.chat.id, "Вы, не можете добавлять записи в базу.")


@dp.message_handler(commands=["gb"])
async def process_photo_command(message: types.Message):
	await bot.send_message(message.chat.id, await send_base())

if __name__ == '__main__':
	bd()
	executor.start_polling(dp)

Сначала нам необходимо создать бота, думаю это может сделать любой и с этим проблем не возникнет. Но на всякий случай скажу что создавать бота необходимо через @BotFather .

Далее мы делаем обработчик на входящие сообщения, а если быть точнее на сообщение с командой "ad" , для меня эта команда значит add то есть добавить. У нас идёт стандартная обработка команды, Далее мы сверяем, если юзер айди равен нашему то - распарсить всё после /ad и вызвать функцию в которую мы будем передавать то что было после /ad. Затем отправлять сообщения что запись успешно добавлена. Если же это пишет какой либо другой пользователь то он не сможет добавить запись в нашу базу данных. Получить свой айди вы можете по средствам бота @get_id_bot

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

Ну и в конце мы создаем главную функцию в которой мы создаем базу данных а затем с помощью метода long_pool получаем сообщения с телеграмма. Выглядит она так:

if __name__ == '__main__':
	bd()
	executor.start_polling(dp)

И да, она была выше)

Вот полный листинг программы:

import sqlite3
import re
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor



def bd():
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	cur.execute('CREATE TABLE IF NOT EXISTS links(Link TEXT)')




async def link_writer(link):
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	mass = []
	mass.append(link)
	cur.execute('INSERT INTO links VALUES(?)', mass)
	con.commit()
	cur.close()


async def send_base():
	con = sqlite3.connect("link.db")
	cur = con.cursor()
	query = 'SELECT * FROM links'
	cur.execute(query)
	data = cur.fetchall()
	mm = []
	for i in data:
		mm.append(i)
	ww = len(data)
	g = []
	for i in range(ww):
		a = re.sub('|\(|\'|\,|\)', '', str(mm[i]))
		g.append(a)
	c = []
	for i in g:
		q = i + "\n"
		c.append(q)
	val = '\n'.join(c)
	return val



TOKEN = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"

bot = Bot(token=TOKEN)
dp = Dispatcher(bot)


@dp.message_handler(commands=['ad'])
async def process_start_command(message: types.Message):
	if message.from_user.id == 11111111111111:
		myString = ' '.join(message.text.split(" ")[1:])
		await link_writer(myString)
		await bot.send_message(message.chat.id, "Запись успешно добавлена в базу.")
	else:
		await bot.send_message(message.chat.id, "Ой, Недостаточно прав.")


@dp.message_handler(commands=["gb"])
async def process_photo_command(message: types.Message):
	await bot.send_message(message.chat.id, await send_base())

if __name__ == '__main__':
	bd()
	executor.start_polling(dp)

Мой код не идеален, вы всегда можете его модернизировать и вообще не допускать тех же ошибок что допускаю я.