Simple Telegram Bot

Решил описать создание простенького телеграмм бота, трудоности с которыми я столкнулся и тому подобнее.

Проект: TempMailBot

Планирую использовать:

  • Python
  • Aiogram
  • Django
  • Celery
  • Redis

Планирую реализовать:

  • Отправку емайлов.
  • Отправку емайлов в будущее.
  • Создание временного емайла.
  • Чтение сообщений с временного емайла
  • Api для пользования временным емайлом

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

Создал Django проект и набросал первые модели, это User, пользователь нашего телеграмм бота, и TempMail, временный емайл пользователя.

from django.db import models


class User(models.Model):
    user_id = models.IntegerField(primary_key=True)

    is_banned = models.BooleanField(default=False)

    is_admin = models.BooleanField(default=False)
    is_moderator = models.BooleanField(default=False)

    created_at  = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return self.user_id

    @classmethod
    def get_user_or_created(cls,update):
        pass



class TempEmail(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)

    email = models.CharField(max_length=255)
    domain = models.CharField(max_length=255)

    created_at = models.DateTimeField(auto_now=True)

    @classmethod
    def del_temp_mail(cls,id):
        cls.objects.filter(
            id=id
        ).delete()


    @classmethod
    @sync_to_async
    def get_temp_mail(cls,user_id):
        e = cls.objects.get(
            user=user_id
        )
        return e

    @classmethod
    @sync_to_async
    def create_temp_mail(cls,u,email,domain=None):
        cls.objects.create(user_id=u,email=email,domain=domain)


Структура проекта -

d

Следующий шаг это создание самого бота и функции отправки сообщения

Создал файл run_pooling, с этого файла будет идти запуск бота, указал какие настройки использовать и загрузил их, ну и путь к нашей функции run_pooling

PS: Хотя можно было создать Managemant commands, не уверен.

import os, django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mailbot.settings')
django.setup()

from mail.handlers.dispatcher import run_pooling

if __name__ == "__main__":
    run_pooling()

bot.py

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

from aiogram import (
    filters,executor,
    types,Bot,Dispatcher
)
from .handlers import send_email_in_the_future

def register_handlers(dp):
  	"""Register handlers"""
    dp.register_message_handler(send_email,commands=['send_mail'])
    # Will be more handlers
    return dp

def run_pooling():
    """Run bot in pooling mod"""
    print("Bot started")
    bot = Bot(token='asd')
    dp = register_handlers(dp=Dispatcher(bot))
    executor.start_polling(dispatcher=dp,skip_updates=True)

Установил библиотеку python-dotenv(слышал что плохая вещь, но ничего сказать не могу), через неё работаю с переменными окружения.Ну и решил использовать джанговскую обертку над smptlib.

Django Settings

load_dotenv()

TOKEN = os.getenv('TOKEN')
EMAIL_HOST = os.getenv('EMAIL_HOST')
EMAIL_HOST_USER = os.getenv('EMAIL_HOST_USER')
EMAIL_HOST_PASSWORD = os.getenv('EMAIL_HOST_PASSWORD')
EMAIL_PORT = os.getenv('EMAIL_PORT')
EMAIL_USE_TLS = os.getenv('EMAIL_USE_TLS')

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

Мой register_handlers показанный выше был обновлен, добавил хендлеры на состояние

def register_handlers(dp):
  	"""Register handlers"""
    dp.register_message_handler(send_email=['send_mail'],state='*')'*')
    dp.register_message_handler(process_subject,state=SendMailForm.subject)
    dp.register_message_handler(process_message,state=SendMailForm.message)
    dp.register_message_handler(process_email,state=SendMailForm.to_mail)
    return dp

И вот обновленный handlers.py

class SendMailForm(StatesGroup):
    subject = State()
    message = State()
    to_mail = State()


async def send_email(m: Message):
    await SendMailForm.subject.set()
    await m.reply(subject)


async def process_subject(message: Message, state: FSMContext):
    async with state.proxy() as data:
        data['subject'] = message.text
    await SendMailForm.next()
    await message.reply("Enter a text to send")


async def process_message(message:Message,state:FSMContext):
    async with state.proxy() as data:
        data['message'] = message.text
    await SendMailForm.next()
    await message.reply("Enter an email")


async def process_email(message:Message,state:FSMContext):
    async with state.proxy() as data:
        data['email'] = message.text
        if not(check(data['email'])):
            await message.reply("Email is not valid,try again from begin")
            await state.finish()


        send_mail(
            data['subject'],
            data['message'],
            'your email',
            [data['email']],
            fail_silently=False,
        )
        await message.answer('Message is sent')
        await state.finish()


Так-же была добавлена простая валидация, находится она в utils.py

import re


regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'

def check(email):
    return True if re.fullmatch(regex,email) else False

Cоздал файл celery.py в настройках проекта. Cамостоятельно Celery не умеет реализовывать периодические задачи, поэтому существует django-celery-beat, не забываем его установить и прокинуть в installed-apps.

import os
from celery import Celery
from datetime import timedelta
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mailbot.settings')

app = Celery('mailbot')
app.config_from_object('django.conf:settings', namespace="CELERY")
app.autodiscover_tasks()

app.conf.update(
    result_expires=3600,
    enable_utc = True,
    timezone = 'UTC'
)

app.conf.beat_schedule = {
    "see-you-in-ten-seconds-task": {
        "task": 'mail.tasks.send_email',
        "schedule":timedelta(seconds=10) # Поставил чисто для себя 
}

Добавил в settings брокера селери и бекенд. Брокер овечает за передачу данных, бекенд за хранение данных.

CELERY_BROKER_URL = 'redis://localhost:6379/0' 
CELERY_RESULT_BACKEND = 'django-db' 

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

mail/handlers/api.py

import requests
import random


class TempMail():
    """
    Api wrapper provides temporary email address
    :param login: (optimal) login for email address
    :param domain:(optimal) domain for email address
    Default domain 1secmail.com
    """

    def __init__(self, login=None, domain='1secmail.com'):
        self.login = login
        self.domain = domain

    def generate_random_email_address(self) -> None:
        """Generates random email"""
        r = requests.get('https://www.1secmail.com/api/v1/?action=genRandomMailbox&count=10')
        get_random = f'{random.choice(r.json())}'
        self.login, self.domain = get_random.split('@')

    @property
    def get_list_of_active_domains(self):
        """Return active domains for email address"""
        return requests.get('https://www.1secmail.com/api/v1/?action=getDomainList').json()

    def get_list_of_emails(self):
        """checks the mailbox for messages and returns them"""
        if self.login is None or self.domain is None:
            self.generate_random_email_address()
        r = requests.get(f'https://www.1secmail.com/api/v1/?action=getMessages&login={self.login}&domain={self.domain}')
        return r.json()

    def get_login(self):
        """Get currently login"""
        return self.login

    def get_domain(self):
        """Get currently domain"""
        return self.domain

    def download_attachment_by_id(self, attachment: str, id: str):
        """
        Downloads attachment from email
        :param attachment:Name of file, example: file1.jpg or file1.png or file1.pdf
        :param id:Id of messages
        """
        if self.login is None or self.domain is None:
            return 'You cant download anything, your login or domain is None.'

        r = requests.get(f"https://www.1secmail.com/api/v1/?action=download&login="
                         f"{self.login}&domain={self.domain}&id={id}&file={attachment}")

        if 'Message not found' in r.text:
            return 'The file could not be found, please check the correctness of the entered data'

        with open(attachment, 'wb') as file:
            file.write(r.content)

    def download_all_files(self):
        """Download all files from mailbox"""
        emails = self.get_list_of_emails()
        lst_with_files = []
        for i in emails:
            r = requests.get(
                f'https://www.1secmail.com/api/v1/?action=readMessage&login={self.login}&domain={self.domain}&id={i["id"]}').json()
            lst_with_files.append(
                {
                    'filename': f"{r['attachments'][0]['filename']}",
                    'id': f"{i['id']}"
                }
            )
        for i in lst_with_files:
            self.download_attachments_by_id(i['filename'], i['id'])

Копируем предыдущую форму по отправке емайла и добавляем date, время в которое будет отправлен емайл. Знаю что по принципу DRY мы не должны повторяться, но не смог додуматься как сделать по другому.

Добавленный участок:

Валидируем емайл, если валиден создаем емайл на отсылку, если нет сообщаем пользователю об этом

async def process_date(message: Message, state: FSMContext):
    async with state.proxy() as data:
        data['date'] = message.text
        day, month, year = message.text.split('-')

        is_valid = True
        try:
            datetime.datetime(int(year), int(month), int(day))
        except ValueError:
            is_valid = False
        if is_valid:
            u,c = await User.get_user_or_created(message.from_user.id)
            await Email.get_email_or_created(u=u,data=data,)
            await message.reply(f'Message will be sent in{year}-{month}-{day}')
        else:
            await message.reply('Date is invalid - format dd-mm-yy')
            await state.finish()

Я решил не добавлять поля EmailField и DataField потому что не вижу в этом смысла, нам уже в БД будет идти валидная дата.

class Email(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)

    subject = models.CharField(max_length=255)

    mail = models.EmailField(max_length=255)

    message = models.CharField(max_length=500)
    date = models.CharField(max_length=255)

    @classmethod
    @sync_to_async
    def get_email_or_created(cls, data, u):
        e, created = cls.objects.get_or_create(
            user=u,
            subject=data['subject'],
            message=data['message'],
            date=data['date'],
            mail=data['email']
        )
        return e, created

    @classmethod
    def get_all_emails(cls):
        return cls.objects.all()

    @classmethod
    def delete_email(cls,id):
        cls.objects.filter(id=id).delete()

    @classmethod
    @sync_to_async
    def get_email_of_user(cls,u):
        e = cls.objects.get(
            user_id=u,
        )
        return e

Столкнулся с первой ошибкой, Celery получает task, но не выполняет его, долго думал в чем причина, гуглил, многие пишут что при переводе проекта с одной версии на другую он у них слетает, скорее всего из-за несовместимсоти чего-то, но вряд-ли у меня этот кейc.

Что только не пробовал сделать, зашел уже в свой старый небольшой проект и решил скопировать настройки, все сделал в точь точь как там, стало ещё хуже, теперь просто висит Starting без какой-либо ошибки. Я потратил на эту ошибку больше времени чем написание всего кода выше, а проблема оказалась в том что я не правильно запускал воркер, нужно было так:

celery-A mailbot worker -l INFO --pool=solo

Так-то суть использования Celery довольно проста и понятна, вопрос лишь в опыте использования данной штуковины, да и если бы я был способен, то с радостью бы залез к ним в исходники, но увы, уровень пока не тот.

Наша таска по отправке сообщений, проверяем если текущая дата совпадает с той которую нужно отправить, то отправляем и после удаляем.

tasks.py

from celery import shared_task
from .models import Email
from datetime import datetime
from django.core.mail import send_mail
@shared_task()
def send_email():
    emails = Email.get_all_emails()
    for i in emails:
        date = i.date
        d,m,y = date.split('-')
        today = datetime.now()
        if datetime(year=today.year,month=today.month,day=today.day) == datetime(year=int(y),month=int(m),day=int(d)):
            send_mail(
                i.subject,
                i.message,
                'your',
                [i.mail],
                fail_silently=False,
            )
            Email.delete_email(id=i.id)

Создаем кнопки для показа всех сообщений в временном емайле -

def make_keyboard_for_messages(emails):
    temp = TempMail()
    temp.domain = emails.domain
    temp.login = emails.email
    lists = temp.get_list_of_emails()
    markup = InlineKeyboardMarkup()
    for i in range(len(lists)):
        markup.add((InlineKeyboardButton(f"{lists[i]['subject']}",callback_data=f'{i}')))
    return markup

Обновленный handlers.py

Добавил форму для создания временного емайла, существовать он будет 30 минут, раз в 30 минут будет запускаться таск для изжитых емайлов.

class CreateTempMail(StatesGroup):
    name = State()
    domain = State()

    
async def create_temp_mail(m:Message):
    await CreateTempMail.name.set()
    await User.get_user_or_created(m.from_user.id)
    await m.reply('Enter name of your email')

    
async def process_email_name(m:Message,state:FSMContext):
        async with state.proxy() as data:
            data['email'] = m.text
            await m.reply(f'Input domain, avaivable domains:\n{t.get_list_of_active_domains}\n'
                          f'if you wont choose, domain will be generated')
            await CreateTempMail.next()


async def process_email_domain(m:Message, state:FSMContext):
    async with state.proxy() as data:
        data['domain'] = m.text
        u,created = await User.get_user_or_created(m.from_user.id)
        await TempEmail.create_temp_mail(u=u,domain=data['domain'],email=data['email'])
        await state.finish()


async def choose_messages_from_temp_mail(m:Message):
    u, created = await User.get_user_or_created(m.from_user.id)
    e = await TempEmail.get_temp_mail(u)
    object = make_keyboard_for_messages(e)
    await m.reply(tex'Choose message you want to read',reply_markup=object)

async def read_messages_from_temp_mail(q:CallbackQuery):
    u,c = await User.get_user_or_created(q.from_user.id)
    e = await TempEmail.get_temp_mail(u)
    t = TempMail(login=e.email,domain=e.domain)
    data = t.get_list_of_emails()[int(q.data)]
    m = t.read_message(id=data['id']).json()
    await q.message.answer(text=f"from: {m['from']}\n subject: {m['subject']}\n text: {m['textBody']}")

Если кому-то вдруг нужен стажер с горящими глазами, то было бы круто.

Вот и все, думаю не стоит показывать как я просто добавил пару хендлеров в bot.py, планировал завернуть все в Docker, написать тест кейсы в Jira, но стало лень, сделал то что сделал.

👍НравитсяПонравилось4
В избранноеВ избранном3
LinkedIn
Допустимые теги: blockquote, a, pre, code, ul, ol, li, b, i, del.
Ctrl + Enter
Допустимые теги: blockquote, a, pre, code, ul, ol, li, b, i, del.
Ctrl + Enter

Що таке у вас «тимчасовий email»? В чому його суть і для чого він може бути потрібний?

Для многократной регистрации на каком-то сайте без использования своего емайла.

То як це працює можете розповісти? У вас буде розгорнутий свій поштовий сервер із своїми доменом?

А, в этом плане, я использую api которая предоставляет эту возможность(неограниченное кол-во запросов) + бесплатная. Сверху я как раз для неё обертку написал

Прикольно. Так ви навіть не задаєте, в якому домені створюються тимчасові адреси?
Я намагаюсь зрозуміти, як взагалі такий сервіс (через який можна розсилати пошту, а потім видаляти адресу, створювати нову і знову розсилати пошту) може існувати, його ж мають банити всі антиспам фільтри.

На таких сайтах как инста, париматч и т.д да, стоит защита от такого, но на некоторых сайтах довольно таки популярных проскакивает, где как. Да, домен выбираем, как это происходит — делаем запрос с логином емайла, создается default домен 1secmail.com , его можно сменить на любой из предложенных. Логин должен желательно быть надежен, так как любой может получить доступ к почте зная его. После того как создали временный аккаунт на час читаем через апишку в которую мы передаем наш логин почту и все.

Дякую, почитав.
Задумка благородна.
Цікаво, що такий сервіс може буде популярним й для накрутки голосів в голосуваннях, де треба підтверджувати себе через e-mail. Або для масової реєстрації бот-акаунтів для ботоферм, або ще щось таке автоматизоване, коли логін-пароль створенного акаунту потім точно зберігається.

капча легко обходится, мб как-то об этом напишу

все что надо знать о современном процессе найма)

уже сколько оферов получил?)

Да не, ошибаешься, такое много оферов не принесет. Здесь есть смысл искать себе работников только мидл + уровня, джуны сами налетят на первую попавшуюся вакансию. Фиг пробьешся на работу, пытаюсь хотя-бы так.

Если кому-то вдруг нужен стажер с горящими глазами, то было бы круто.

написал сообщение

Проверил почту, сообщение не пришло.
Мой телеграмм — @krenik
Я сейчас вам ещё на почту отпишу, вдруг от меня придет и вы мне просто ответите там.

Оставить гитхаб о котором я забыл))
github.com/leirons/MailBot

Подписаться на комментарии