from constructor_telegram_bots.celery import celery_app


__all__ = ('celery_app',)

constructor_telegram_bots/constructor_telegram_bots/__init__.py
pyguru-dev/telegrambot
from django.core.asgi import get_asgi_application

import os


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'constructor_telegram_bots.settings')
application = get_asgi_application()

constructor_telegram_bots/constructor_telegram_bots/asgi.py
pyguru-dev/telegrambot
from celery import Celery, signals

import os


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'constructor_telegram_bots.settings')


celery_app = Celery('constructor_telegram_bots')
celery_app.config_from_object('django.conf:settings', namespace='CELERY')
celery_app.autodiscover_tasks(['user', 'telegram_bot.services'])


@signals.celeryd_after_setup.connect
def celery_after_setup(*args, **kwargs):
	from telegram_bot.services.tasks import start_all_telegram_bots
	start_all_telegram_bots.delay()

constructor_telegram_bots/constructor_telegram_bots/celery.py
pyguru-dev/telegrambot
from django.conf import settings


def constructor_telegram_bot_username(*args, **kwargs) -> dict:
    return {'constructor_telegram_bot_username': settings.CONSTRUCTOR_TELEGRAM_BOT_USERNAME}

constructor_telegram_bots/constructor_telegram_bots/context_processors.py
pyguru-dev/telegrambot
from django.conf import settings

from asgiref.sync import sync_to_async
import requests
import jinja2


def create_user(user) -> None:
	if settings.DEBUG_ENVIRONMENT or settings.TEST:
		return

	requests.post('http://127.0.0.1:99/users/', json={'user_id': user.id, 'token': user.auth_token.key})

def delete_user(user) -> None:
	if settings.DEBUG_ENVIRONMENT or settings.TEST:
		return

	requests.delete(f'http://127.0.0.1:99/users/{user.id}/')

def add_plugin(plugin) -> None:
	if settings.DEBUG_ENVIRONMENT or settings.TEST:
		return

	requests.post('http://127.0.0.1:99/plugins/', json={
		'user_id': plugin.user.id,
		'token': plugin.user.auth_token.key,
		'telegram_bot_id': plugin.telegram_bot.id,
		'plugin_id': plugin.id,
		'name': plugin.name,
		'code': plugin.code,
	})

def update_plugin(plugin) -> None:
	if settings.DEBUG_ENVIRONMENT or settings.TEST:
		return

	requests.post('http://127.0.0.1:99/plugins/', json={
		'user_id': plugin.user.id,
		'token': plugin.user.auth_token.key,
		'telegram_bot_id': plugin.telegram_bot.id,
		'plugin_id': plugin.id,
		'name': plugin.name,
		'code': plugin.code,
	})

def delete_plugin(plugin) -> None:
	if settings.DEBUG_ENVIRONMENT or settings.TEST:
		return

	requests.delete('http://127.0.0.1:99/plugins/', json={
		'user_id': plugin.user.id,
		'telegram_bot_id': plugin.telegram_bot.id,
		'name': plugin.name,
	})

def replace_text_variables(telegram_bot, text: str, text_variables: dict) -> str:
	if settings.DEBUG_ENVIRONMENT or settings.TEST:
		environment = jinja2.Environment()
		template: jinja2.Template = environment.from_string(text)
		text: str = template.render(text_variables)
		return text.replace('\n\n', '\n')

	return requests.post('http://127.0.0.1:99/generate/template/', json={
		'user_id': telegram_bot.owner.id,
		'token': telegram_bot.owner.auth_token.key,
		'telegram_bot_id': telegram_bot.id,
		'text': text,
		'text_variables': text_variables,
	}).json()['response']

async def areplace_text_variables(telegram_bot, text: str, text_variables: dict) -> str:
	return await sync_to_async(replace_text_variables)(telegram_bot=telegram_bot, text=text, text_variables=text_variables)

constructor_telegram_bots/constructor_telegram_bots/environment.py
pyguru-dev/telegrambot
from drf_standardized_errors.formatter import ExceptionFormatter
from drf_standardized_errors.types import ErrorResponse, Error


class CustomExceptionFormatter(ExceptionFormatter):
	def format_error_response(self, error_response: ErrorResponse) -> dict:
		error: Error = error_response.errors[0]

		return {
			'code': error.code,
			'name': error.attr,
			'message': error.detail,
			'level': 'danger',
		}

constructor_telegram_bots/constructor_telegram_bots/exception_formatter.py
pyguru-dev/telegrambot
import random


def generate_random_string(length: int, chars: str) -> str:
	return ''.join([random.choice(chars) for num in range(length)])

constructor_telegram_bots/constructor_telegram_bots/functions.py
pyguru-dev/telegrambot
from constructor_telegram_bots.settings import BASE_DIR


workers = 9
max_requests = 1000
max_requests_jitter = 100


capture_output = True
accesslog = str(BASE_DIR / 'logs/gunicorn_info.log')
errorlog = str(BASE_DIR / 'logs/gunicorn_info.log')

constructor_telegram_bots/constructor_telegram_bots/gunicorn_config.py
pyguru-dev/telegrambot
from django.utils.translation import gettext_lazy as _

from .functions import generate_random_string

from dotenv import load_dotenv
from pathlib import Path
import sys
import os


BASE_DIR = Path(__file__).resolve().parent.parent


load_dotenv()

SECRET_KEY = os.getenv('SECRET_KEY', f"django-insecure-{generate_random_string(length=50, chars='abcdefghijklnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890!@#$%^&*()_')}")
DEBUG = os.getenv('DEBUG', 'True') == 'True'
DEBUG_ENVIRONMENT = os.getenv('DEBUG_ENVIRONMENT', 'True') == 'True'
TEST = len(sys.argv) >= 2 and sys.argv[0] == 'manage.py' and sys.argv[1] == 'test'

CONSTRUCTOR_TELEGRAM_BOT_API_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
CONSTRUCTOR_TELEGRAM_BOT_USERNAME = os.getenv('TELEGRAM_BOT_USERNAME')


SITE_DOMAIN = 'http://127.0.0.1:8000' if DEBUG else 'https://constructor.exg1o.org'
ALLOWED_HOSTS = ['127.0.0.1', 'constructor.exg1o.org']
CSRF_TRUSTED_ORIGINS = [
	'http://*.127.0.0.1',
	'https://*.127.0.0.1',
	'http://constructor.exg1o.org',
	'https://constructor.exg1o.org',
]


CELERY_BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_BEAT_SCHEDULE = {
	'check_users_first_name_schedule' : {
		'task': 'user.tasks.check_users_first_name',
		'schedule': 86400, # 60 * 60 * 24 = 86400 секунд
	},
}


INSTALLED_APPS = [
	'rest_framework',
	'rest_framework.authtoken',
	'drf_standardized_errors',

	'admin_interface',
	'colorfield',

	'modeltranslation',
	'django_json_widget',
	'tinymce',

	'django.contrib.admin',
	'django.contrib.auth',
	'django.contrib.contenttypes',
	'django.contrib.sessions',
	'django.contrib.messages',
	'django.contrib.staticfiles',

	'user',
	'home',
	'team',
	'updates',
	'instruction',
	'donation',
	'personal_cabinet',
	'telegram_bot_menu',
	'telegram_bot',
	'plugin',
	'privacy_policy',
]

MIDDLEWARE = [
	'django.middleware.security.SecurityMiddleware',
	'django.contrib.sessions.middleware.SessionMiddleware',
	'django.middleware.common.CommonMiddleware',
	'django.middleware.csrf.CsrfViewMiddleware',
	'django.contrib.auth.middleware.AuthenticationMiddleware',
	'django.contrib.messages.middleware.MessageMiddleware',
	'django.middleware.clickjacking.XFrameOptionsMiddleware',
	'django.middleware.locale.LocaleMiddleware',
]

REST_FRAMEWORK = {
	'DEFAULT_AUTHENTICATION_CLASSES': [
		'rest_framework.authentication.BasicAuthentication',
		'rest_framework.authentication.SessionAuthentication',
	],
	'EXCEPTION_HANDLER': 'drf_standardized_errors.handler.exception_handler',
}
DRF_STANDARDIZED_ERRORS = {
	'EXCEPTION_FORMATTER_CLASS': 'constructor_telegram_bots.exception_formatter.CustomExceptionFormatter',
}

TINYMCE_DEFAULT_CONFIG = {
	'theme': 'silver',
	'height': 420,
	'menubar': True,
	'plugins': 'advlist, autolink, lists, link, image, charmap, print, preview, anchor, searchreplace, visualblocks, code, fullscreen, insertdatetime, media, table, paste, code, help, wordcount',
	'toolbar': 'undo redo | formatselect | bold italic backcolor | alignleft aligncenter alignright alignjustify | bullist numlist outdent indent | removeformat | help',
}

TEMPLATES = [
	{
		'BACKEND': 'django.template.backends.django.DjangoTemplates',
		'DIRS': [
			BASE_DIR / 'templates',
			BASE_DIR / 'constructor_telegram_bots/templates',
		],
		'APP_DIRS': True,
		'OPTIONS': {
			'context_processors': [
				'django.template.context_processors.debug',
				'django.template.context_processors.request',
				'django.contrib.auth.context_processors.auth',
				'django.contrib.messages.context_processors.messages',

				'constructor_telegram_bots.context_processors.constructor_telegram_bot_username',
				'user.context_processors.users',
				'telegram_bot.context_processors.telegram_bots',
				'team.context_processors.team_members',
				'updates.context_processors.updates',
				'instruction.context_processors.instruction_sections',
				'donation.context_processors.donations',
				'donation.context_processors.donation_sections',
			],
		},
	}
]

ROOT_URLCONF = 'constructor_telegram_bots.urls'

WSGI_APPLICATION = 'constructor_telegram_bots.wsgi.application'


AUTH_USER_MODEL = 'user.User'

if TEST:
	DATABASES = {
		'default': {
			'ENGINE': 'django.db.backends.sqlite3',
			'NAME': BASE_DIR / 'DataBase.db',
		},
	}
else:
	DATABASES = {
		'default': {
			'ENGINE': 'django.db.backends.postgresql',
			'NAME': os.getenv('POSTGRESQL_DATABASE_NAME'), 
			'USER': os.getenv('POSTGRESQL_DATABASE_USER'),
			'PASSWORD': os.getenv('POSTGRESQL_DATABASE_PASSWORD'),
			'HOST': '127.0.0.1', 
			'PORT': '5432',
		},
	}

DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'


USE_I18N = True
USE_L10N = True

LANGUAGE_CODE = 'ru-ru'
LANGUAGES = (
	('en', _('Английский')),
	('uk', _('Украинский')),
	('ru', _('Русский')),
)
MODELTRANSLATION_DEFAULT_LANGUAGE = 'ru'

LOCALE_PATHS = (BASE_DIR / 'locale',)


TIME_ZONE = 'UTC'
USE_TZ = True


STATIC_URL = '/static/'
STATIC_ROOT = BASE_DIR / 'static'
STATICFILES_DIRS = [BASE_DIR / 'constructor_telegram_bots/static']

MEDIA_URL = '/media/'
MEDIA_ROOT = BASE_DIR / 'media'


if os.path.exists(BASE_DIR / 'logs') is False:
	os.mkdir(BASE_DIR / 'logs')

LOGGING = {
	'version': 1,
	'disable_existing_loggers': False,
	'formatters': {
		'verbose': {
			'format': '[{asctime}]: {levelname}: {name} > {funcName} || {message}',
			'style': '{',
		},
		'simple': {
			'format': '[{asctime}]: {message}',
			'style': '{',
		},
	},
	'handlers': {
		'console': {
			'level': 'INFO',
			'class': 'logging.StreamHandler',
			'formatter': 'simple',
		},
		'django_info_file': { 
			'level': 'DEBUG',
			'class': 'logging.handlers.RotatingFileHandler',
			'filename': BASE_DIR / 'logs/django_info.log',
			'maxBytes': 10485760,
			'backupCount': 10,
			'formatter': 'verbose',
		},
		'django_error_file': { 
			'level': 'WARNING',
			'class': 'logging.handlers.RotatingFileHandler',
			'filename': BASE_DIR / 'logs/django_error.log',
			'maxBytes': 10485760,
			'backupCount': 10,
			'formatter': 'verbose',
		},
		'telegram_bots_info_file': {
			'level': 'INFO',
			'class': 'logging.handlers.RotatingFileHandler',
			'filename': BASE_DIR / 'logs/telegram_bots_info.log',
			'maxBytes': 10485760,
			'backupCount': 10,
			'formatter': 'verbose',
		},
		'telegram_bots_error_file': {
			'level': 'WARNING',
			'class': 'logging.handlers.RotatingFileHandler',
			'filename': BASE_DIR / 'logs/telegram_bots_error.log',
			'maxBytes': 10485760,
			'backupCount': 10,
			'formatter': 'verbose',
		},
	},
	'loggers': {
		'django': {
			'handlers': [
				'console',
				'django_info_file',
				'django_error_file',
			],
			'propagate': True,
		},
		'aiogram': {
			'handlers': [
				'telegram_bots_info_file',
				'telegram_bots_error_file',
			],
			'propagate': True,
		},
	},
}

constructor_telegram_bots/constructor_telegram_bots/settings.py
pyguru-dev/telegrambot
from django.urls import path, include
from django.contrib import admin
from django.conf import settings

from telegram_bot.services import tasks

import sys


urlpatterns = [
	path('admin/', admin.site.urls),
	path('i18n/', include('django.conf.urls.i18n')),

	path('tinymce/', include('tinymce.urls')),

	path('user/', include('user.urls')),
	path('', include('home.urls')),
	path('team/', include('team.urls')),
	path('updates/', include('updates.urls')),
	path('instruction/', include('instruction.urls')),
	path('donation/', include('donation.urls')),
	path('personal-cabinet/', include('personal_cabinet.urls')),
	path('telegram-bot-menu/', include('telegram_bot_menu.urls')),
	path('telegram-bots/', include('telegram_bot.urls')),
	path('plugins/', include('plugin.urls')),
	path('privacy-policy/', include('privacy_policy.urls')),
]


if settings.DEBUG:
	from django.conf.urls.static import static

	urlpatterns += static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
	urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

if not settings.TEST and sys.argv[1] == 'runserver' and sys.platform == 'win32':
	tasks.start_all_telegram_bots()

constructor_telegram_bots/constructor_telegram_bots/urls.py
pyguru-dev/telegrambot
from django.core.wsgi import get_wsgi_application

import os


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'constructor_telegram_bots.settings')
application = get_wsgi_application()

constructor_telegram_bots/constructor_telegram_bots/wsgi.py
pyguru-dev/telegrambot
from django.contrib import admin
from django.db import models
from django.utils.translation import gettext_lazy as _
from django.utils import html

from modeltranslation.admin import TranslationAdmin
from tinymce.widgets import TinyMCE

from .models import *


@admin.register(Donation)
class DonationAdmin(admin.ModelAdmin):
	date_hierarchy = 'date'

	list_display = ('id', 'sum_', 'telegram_url_', 'date')
	fields = ('sum', 'telegram_url', 'date')

	@admin.display(description=_('Сумма'), ordering='sum')
	def sum_(self, donation: Donation) -> str:
		return f'{donation.sum}€'

	@admin.display(description=_('Telegram'), ordering='telegram_url')
	def telegram_url_(self, donation: Donation) -> str:
		return html.format_html(f'<a href="{donation.telegram_url}" style="font-weight: 600;" target="_blank">{donation.telegram_url}</a>')

@admin.register(DonationSection)
class DonationSectionAdmin(TranslationAdmin):
	list_display = ('title', 'position')

	fields = ('title', 'text', 'position')
	formfield_overrides = {
		models.TextField: {
			'widget': TinyMCE,
		},
	}

constructor_telegram_bots/donation/admin.py
pyguru-dev/telegrambot
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _


class DonationConfig(AppConfig):
	default_auto_field = 'django.db.models.BigAutoField'

	name = 'donation'
	verbose_name = _('Пожертвования')

constructor_telegram_bots/donation/apps.py
pyguru-dev/telegrambot
from .models import *


def donations(*args, **kwargs) -> dict:
	return {'donations': Donation.objects.all()}

def donation_sections(*args, **kwargs) -> dict:
	return {'donation_sections': DonationSection.objects.all()}

constructor_telegram_bots/donation/context_processors.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-02 01:39

from django.db import migrations, models


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='Donation',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('date', models.DateTimeField(verbose_name='Дата')),
                ('telegram_url', models.CharField(max_length=255, verbose_name='Ссылка на Telegram')),
                ('sum', models.FloatField(verbose_name='Сумма')),
            ],
            options={
                'verbose_name': 'Пожертвование',
                'verbose_name_plural': 'Пожертвования',
                'db_table': 'donation',
            },
        ),
    ]

constructor_telegram_bots/donation/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-08-02 17:04

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('donation', '0001_initial'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='donation',
            options={'ordering': ['-date'], 'verbose_name': 'Пожертвование', 'verbose_name_plural': 'Пожертвования'},
        ),
    ]

constructor_telegram_bots/donation/migrations/0002_alter_donation_options.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 20:16

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('donation', '0002_alter_donation_options'),
    ]

    operations = [
        migrations.CreateModel(
            name='DonationSection',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('position', models.IntegerField(verbose_name='Позиция')),
                ('title', models.CharField(max_length=255, verbose_name='Заголовок')),
                ('title_en', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('title_uk', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('title_ru', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('text', models.TextField(verbose_name='Текст')),
                ('text_en', models.TextField(null=True, verbose_name='Текст')),
                ('text_uk', models.TextField(null=True, verbose_name='Текст')),
                ('text_ru', models.TextField(null=True, verbose_name='Текст')),
            ],
            options={
                'verbose_name': 'Раздел',
                'verbose_name_plural': 'Разделы',
                'db_table': 'donation_section',
                'ordering': ['position'],
            },
        ),
    ]

constructor_telegram_bots/donation/migrations/0003_donationsection.py
pyguru-dev/telegrambot
from django.db import models
from django.utils.translation import gettext_lazy as _


class Donation(models.Model):
	sum = models.FloatField(_('Сумма'))
	telegram_url = models.CharField(_('Ссылка на Telegram'), max_length=255)
	date = models.DateTimeField(_('Дата'))

	class Meta:
		db_table = 'donation'
		ordering = ['-date']

		verbose_name = _('Пожертвование')
		verbose_name_plural = _('Пожертвования')

	def __str__(self) -> str:
		return self.telegram_url

class DonationSection(models.Model):
	title = models.CharField(_('Заголовок'), max_length=255)
	text = models.TextField(_('Текст'))
	position = models.IntegerField(_('Позиция'))

	class Meta:
		db_table = 'donation_section'
		ordering = ['position']

		verbose_name = _('Раздел')
		verbose_name_plural = _('Разделы')

	def __str__(self) -> str:
		return self.title
constructor_telegram_bots/donation/models.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls

from .models import *

from datetime import datetime


class DonationModelTests(TestCase):
	def setUp(self) -> None:
		self.current_date: datetime = datetime.now()
		self.donation: Donation = Donation.objects.create(
			sum=50.00,
			telegram_url='https://example.com/',
			date=self.current_date
		)

	def test_fields(self) -> None:
		self.assertEqual(self.donation.sum, 50.00)
		self.assertEqual(self.donation.telegram_url, 'https://example.com/')
		self.assertEqual(self.donation.date, self.current_date)

class DonationSectionModelTests(TestCase):
	def setUp(self) -> None:
		self.current_date: datetime = datetime.now()
		self.donation_section: DonationSection = DonationSection.objects.create(
			title='Test',
			text='Test...',
			position=1
		)

	def test_fields(self) -> None:
		self.assertEqual(self.donation_section.title, 'Test')
		self.assertEqual(self.donation_section.text, 'Test...')
		self.assertEqual(self.donation_section.position, 1)

class ViewsTests(TestCase):
	def test_donation_view(self) -> None:
		url: str = urls.reverse('donation')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'donation.html')

	def test_donation_completed_view(self) -> None:
		url: str = urls.reverse('donation_completed')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'donation_completed.html')

constructor_telegram_bots/donation/tests.py
pyguru-dev/telegrambot
from modeltranslation.translator import register, TranslationOptions

from .models import DonationSection


@register(DonationSection)
class DonationSectionTranslationOptions(TranslationOptions):
    fields = ('title', 'text')

constructor_telegram_bots/donation/translation.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.donation_view, name='donation'),
    path('completed/', views.donation_completed_view, name='donation_completed'),
]

constructor_telegram_bots/donation/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render


def donation_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'donation.html')

def donation_completed_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'donation_completed.html')

constructor_telegram_bots/donation/views.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls


class ViewsTests(TestCase):
	def test_home_view(self) -> None:
		url: str = urls.reverse('home')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'home.html')

constructor_telegram_bots/home/tests.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render


def home_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'home.html')

constructor_telegram_bots/home/views.py
pyguru-dev/telegrambot
from django.contrib import admin

from modeltranslation.admin import TranslationAdmin
from tinymce.widgets import TinyMCE

from django.db import models

from .models import InstructionSection


@admin.register(InstructionSection)
class InstructionSectionAdmin(TranslationAdmin):
	list_display = ('title', 'position')

	fields = ('title', 'text', 'position')
	formfield_overrides = {
		models.TextField: {
			'widget': TinyMCE,
		},
	}

constructor_telegram_bots/instruction/admin.py
pyguru-dev/telegrambot
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _


class InstructionConfig(AppConfig):
	default_auto_field = 'django.db.models.BigAutoField'

	name = 'instruction'
	verbose_name = _('Инструкция')

constructor_telegram_bots/instruction/apps.py
pyguru-dev/telegrambot
from .models import InstructionSection


def instruction_sections(*args, **kwargs) -> dict:
    return {'instruction_sections': InstructionSection.objects.all()}

constructor_telegram_bots/instruction/context_processors.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-01 20:52

from django.db import migrations, models


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='InstructionSection',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('position', models.IntegerField(verbose_name='Позиция')),
                ('title', models.CharField(max_length=255, verbose_name='Заголовок')),
                ('title_en', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('title_uk', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('title_ru', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('text', models.TextField(verbose_name='Текст')),
                ('text_en', models.TextField(null=True, verbose_name='Текст')),
                ('text_uk', models.TextField(null=True, verbose_name='Текст')),
                ('text_ru', models.TextField(null=True, verbose_name='Текст')),
                ('last_update', models.DateTimeField(auto_now=True, verbose_name='Дата обновления')),
            ],
            options={
                'verbose_name': 'Раздел',
                'verbose_name_plural': 'Разделы',
                'db_table': 'instruction_section',
            },
        ),
    ]

constructor_telegram_bots/instruction/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-01 22:41

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('instruction', '0001_initial'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='instructionsection',
            options={'ordering': ['position'], 'verbose_name': 'Раздел', 'verbose_name_plural': 'Разделы'},
        ),
    ]

constructor_telegram_bots/instruction/migrations/0002_alter_instructionsection_options.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-11 21:24

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('instruction', '0002_alter_instructionsection_options'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='instructionsection',
            options={'ordering': ['position'], 'verbose_name': 'Раздел инструкции', 'verbose_name_plural': 'Разделы инструкции'},
        ),
    ]

constructor_telegram_bots/instruction/migrations/0003_alter_instructionsection_options.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 21:07

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('instruction', '0003_alter_instructionsection_options'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='instructionsection',
            options={'ordering': ['position'], 'verbose_name': 'Раздел', 'verbose_name_plural': 'Разделы'},
        ),
        migrations.RemoveField(
            model_name='instructionsection',
            name='last_update',
        ),
    ]

constructor_telegram_bots/instruction/migrations/0004_alter_instructionsection_options_and_more.py
pyguru-dev/telegrambot
from django.db import models
from django.utils.translation import gettext_lazy as _


class InstructionSection(models.Model):
	title = models.CharField(_('Заголовок'), max_length=255)
	text = models.TextField(_('Текст'))
	position = models.IntegerField(_('Позиция'))

	class Meta:
		db_table = 'instruction_section'
		ordering = ['position']

		verbose_name = _('Раздел')
		verbose_name_plural = _('Разделы')

	def __str__(self) -> str:
		return self.title

constructor_telegram_bots/instruction/models.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls

from .models import *


class InstructionSectionModelTests(TestCase):
	def setUp(self) -> None:
		self.instruction_section: InstructionSection = InstructionSection.objects.create(
			title='Test',
			text='Test...',
			position=1
		)

	def test_fields(self) -> None:
		self.assertEqual(self.instruction_section.title, 'Test')
		self.assertEqual(self.instruction_section.text, 'Test...')
		self.assertEqual(self.instruction_section.position, 1)

class ViewsTests(TestCase):
	def test_instruction_view(self) -> None:
		url: str = urls.reverse('instruction')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'instruction.html')

constructor_telegram_bots/instruction/tests.py
pyguru-dev/telegrambot
from modeltranslation.translator import register, TranslationOptions

from .models import InstructionSection


@register(InstructionSection)
class InstructionSectionTranslationOptions(TranslationOptions):
    fields = ('title', 'text')

constructor_telegram_bots/instruction/translation.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.instruction_view, name='instruction'),
]

constructor_telegram_bots/instruction/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render


def instruction_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'instruction.html')

constructor_telegram_bots/instruction/views.py
pyguru-dev/telegrambot
import sys
import os


def main() -> None:
	os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'constructor_telegram_bots.settings')

	try:
		from django.core.management import execute_from_command_line
	except ImportError as exc:
		raise ImportError(
			"Couldn't import Django. Are you sure it's installed and "
			"available on your PYTHONPATH environment variable? Did you "
			"forget to activate a virtual environment?"
		) from exc

	execute_from_command_line(sys.argv)

if __name__ == '__main__':
	main()

constructor_telegram_bots/manage.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls

from user.models import User


class ViewsTests(TestCase):
	def setUp(self) -> None:
		self.user: User = User.objects.create(telegram_id=123456789, first_name='exg1o')

	def test_personal_cabinet_view(self) -> None:
		url: str = urls.reverse('personal_cabinet')

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'personal_cabinet.html')

constructor_telegram_bots/personal_cabinet/tests.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.personal_cabinet_view, name='personal_cabinet'),
]

constructor_telegram_bots/personal_cabinet/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render

from django.contrib.auth.decorators import login_required


@login_required
def personal_cabinet_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'personal_cabinet.html')

constructor_telegram_bots/personal_cabinet/views.py
pyguru-dev/telegrambot
from django.contrib import admin

from .models import *


@admin.register(Plugin)
class PluginAdmin(admin.ModelAdmin):
	date_hierarchy = 'added_date'
	list_filter = ('is_checked', 'added_date')

	list_display = ('id', 'user', 'telegram_bot', 'name', 'is_checked', 'added_date')
	fields = ('user', 'telegram_bot', 'name', 'code', 'is_checked')

@admin.register(PluginLog)
class PluginLogAdmin(admin.ModelAdmin):
	date_hierarchy = 'added_date'
	list_filter = ('level', 'added_date')

	list_display = ('id', 'user', 'telegram_bot', 'plugin', 'message', 'level', 'added_date')
	fields = ('user', 'telegram_bot', 'plugin', 'message', 'level')

constructor_telegram_bots/plugin/admin.py
pyguru-dev/telegrambot
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _


class PluginConfig(AppConfig):
	default_auto_field = 'django.db.models.BigAutoField'

	name = 'plugin'
	verbose_name = _('Плагины')

constructor_telegram_bots/plugin/apps.py
pyguru-dev/telegrambot
from django.utils.translation import gettext as _

from rest_framework.response import Response
from rest_framework.status import *

from .models import Plugin

from functools import wraps


def check_plugin_id(func):
	@wraps(func)
	def wrapper(*args, **kwargs):
		plugin_id: int = kwargs.pop('plugin_id')

		if not Plugin.objects.filter(id=plugin_id).exists():
			return Response({
				'message': _('Плагин не найден!'),
				'level': 'danger',
			}, status=HTTP_404_NOT_FOUND)

		return func(plugin=Plugin.objects.get(id=plugin_id), *args, **kwargs)
	return wrapper

constructor_telegram_bots/plugin/decorators.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-11 23:34

from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    initial = True

    dependencies = [
        ('telegram_bot', '0014_alter_telegrambotuser_options_and_more'),
        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
    ]

    operations = [
        migrations.CreateModel(
            name='Plugin',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('name', models.CharField(max_length=255, verbose_name='Название')),
                ('code', models.TextField(verbose_name='Код')),
                ('is_checked', models.BooleanField(verbose_name='Проверен')),
                ('date_added', models.DateTimeField(auto_now_add=True, verbose_name='Дата добавления')),
                ('telegram_bot', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='telegram_bot.telegrambot', verbose_name='Telegram бот')),
                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL, verbose_name='Пользователь')),
            ],
            options={
                'verbose_name': 'Плагин',
                'verbose_name_plural': 'Плагины',
                'db_table': 'plugin',
            },
        ),
        migrations.CreateModel(
            name='PluginLog',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('message', models.TextField(verbose_name='Сообщение')),
                ('level', models.CharField(choices=[('info', 'Info'), ('success', 'Success'), ('danger', 'Danger')], default='info', max_length=7, verbose_name='Уровень')),
                ('date_added', models.DateTimeField(auto_now_add=True, verbose_name='Дата добавления')),
                ('plugin', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='plugin.plugin', verbose_name='Плагин')),
                ('telegram_bot', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='telegram_bot.telegrambot', verbose_name='Telegram бот')),
                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL, verbose_name='Пользователь')),
            ],
            options={
                'verbose_name': 'Журнал плагина',
                'verbose_name_plural': 'Журналы плагинов',
                'db_table': 'plugin_log',
            },
        ),
    ]

constructor_telegram_bots/plugin/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-12 01:16

from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        ('plugin', '0001_initial'),
    ]

    operations = [
        migrations.RenameField(
            model_name='plugin',
            old_name='date_added',
            new_name='added_date',
        ),
        migrations.RenameField(
            model_name='pluginlog',
            old_name='date_added',
            new_name='added_date',
        ),
        migrations.AlterField(
            model_name='plugin',
            name='is_checked',
            field=models.BooleanField(default=False, verbose_name='Проверен'),
        ),
        migrations.AlterField(
            model_name='pluginlog',
            name='plugin',
            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='logs', to='plugin.plugin', verbose_name='Плагин'),
        ),
    ]

constructor_telegram_bots/plugin/migrations/0002_rename_date_added_plugin_added_date_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-14 14:07

from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0014_alter_telegrambotuser_options_and_more'),
        ('plugin', '0002_rename_date_added_plugin_added_date_and_more'),
    ]

    operations = [
        migrations.RenameField(
            model_name='plugin',
            old_name='added_date',
            new_name='date_added',
        ),
        migrations.RenameField(
            model_name='pluginlog',
            old_name='added_date',
            new_name='date_added',
        ),
        migrations.AlterField(
            model_name='plugin',
            name='telegram_bot',
            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='plugins', to='telegram_bot.telegrambot', verbose_name='Telegram бот'),
        ),
    ]

constructor_telegram_bots/plugin/migrations/0003_rename_added_date_plugin_date_added_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-28 10:34

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('plugin', '0003_rename_added_date_plugin_date_added_and_more'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='pluginlog',
            options={'verbose_name': 'Логи плагина', 'verbose_name_plural': 'Логи плагинов'},
        ),
    ]

constructor_telegram_bots/plugin/migrations/0004_alter_pluginlog_options.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 21:22

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('plugin', '0004_alter_pluginlog_options'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='pluginlog',
            options={'verbose_name': 'Логи', 'verbose_name_plural': 'Логи'},
        ),
        migrations.RenameField(
            model_name='plugin',
            old_name='date_added',
            new_name='added_date',
        ),
        migrations.RenameField(
            model_name='pluginlog',
            old_name='date_added',
            new_name='added_date',
        ),
    ]

constructor_telegram_bots/plugin/migrations/0005_alter_pluginlog_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 21:23

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('plugin', '0005_alter_pluginlog_options_and_more'),
    ]

    operations = [
        migrations.AlterField(
            model_name='plugin',
            name='added_date',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Добавлен'),
        ),
        migrations.AlterField(
            model_name='pluginlog',
            name='added_date',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Добавлен'),
        ),
    ]

constructor_telegram_bots/plugin/migrations/0006_alter_plugin_added_date_alter_pluginlog_added_date.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-12 20:34

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('plugin', '0006_alter_plugin_added_date_alter_pluginlog_added_date'),
    ]

    operations = [
        migrations.AlterField(
            model_name='pluginlog',
            name='level',
            field=models.CharField(choices=[('info', 'Информация'), ('success', 'Успех'), ('danger', 'Ошибка')], default='info', max_length=7, verbose_name='Уровень'),
        ),
    ]

constructor_telegram_bots/plugin/migrations/0007_alter_pluginlog_level.py
pyguru-dev/telegrambot
from django.db import models
from django.template import defaultfilters as filters
from django.utils.translation import gettext_lazy as _

from user.models import User
from telegram_bot.models import TelegramBot

from constructor_telegram_bots import environment


class Plugin(models.Model):
	user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name=_('Пользователь'))
	telegram_bot = models.ForeignKey(TelegramBot, on_delete=models.CASCADE, related_name='plugins', verbose_name=_('Telegram бот'))
	name = models.CharField(_('Название'), max_length=255)
	code = models.TextField(_('Код'))
	is_checked = models.BooleanField(_('Проверен'), default=False)
	added_date = models.DateTimeField(_('Добавлен'), auto_now_add=True)

	class Meta:
		db_table = 'plugin'

		verbose_name = _('Плагин')
		verbose_name_plural = _('Плагины')

	def to_dict(self) -> dict:
		return {
			'id': self.id,
			'name': self.name,
			'code': self.code,
			'is_checked': self.is_checked,
			'added_date': f'{filters.date(self.added_date)} {filters.time(self.added_date)}',
		}

	def save(self, *args, **kwargs) -> None:
		if self.is_checked:
			environment.update_plugin(self)
		super().save(*args, **kwargs)

	def delete(self, *args, **kwargs) -> None:
		if self.is_checked:
			environment.delete_plugin(self)
		super().delete(*args, **kwargs)

	def __str__(self) -> str:
		return self.name

class PluginLog(models.Model):
	user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name=_('Пользователь'))
	telegram_bot = models.ForeignKey(TelegramBot, on_delete=models.CASCADE, verbose_name=_('Telegram бот'))
	plugin = models.ForeignKey(Plugin, on_delete=models.CASCADE, related_name='logs', verbose_name=_('Плагин'))
	message = models.TextField(_('Сообщение'))
	level = models.CharField(_('Уровень'), max_length=7, choices=(
		('info', _('Информация')),
		('success', _('Успех')),
		('danger', _('Ошибка'))
	), default='info')
	added_date = models.DateTimeField(_('Добавлен'), auto_now_add=True)

	class Meta:
		db_table = 'plugin_log'

		verbose_name = _('Логи')
		verbose_name_plural = _('Логи')

	def to_dict(self) -> dict:
		return {
			'plugin_name': self.plugin.name,
			'message': self.message,
			'level': self.level,
			'added_date': f'{filters.date(self.added_date)} {filters.time(self.added_date)}',
		}

	def __str__(self) -> str:
		return self.plugin.name

constructor_telegram_bots/plugin/models.py
pyguru-dev/telegrambot
from django.utils.translation import gettext_lazy as _

from rest_framework import serializers

from telegram_bot.models import TelegramBot

import re


class CreatePluginSerializer(serializers.Serializer):
	name = serializers.CharField(max_length=255, error_messages={
		'blank': _('Введите название плагина!'),
		'max_length': _('Название плагина должно содержать не более 255 символов!'),
	})
	code = serializers.CharField()

	def validate_name(self, name: str) -> str:
		telegram_bot: TelegramBot = self.context['telegram_bot']

		if re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', name) is None:
			raise serializers.ValidationError(_('Название плагина содержит запрещенные символы!'))

		if telegram_bot.plugins.filter(name=name).exists():
			raise serializers.ValidationError(_('У вас уже добавлен плагин с таким названием!'))

		return name

class UpdatePluginSerializer(serializers.Serializer):
	code = serializers.CharField()

class AddPluginLogSerializer(serializers.Serializer):
	message = serializers.CharField()
	level = serializers.ChoiceField(choices=['info', 'success', 'danger'])

constructor_telegram_bots/plugin/serializers.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls
from django.template import defaultfilters as filters

from user.models import User
from telegram_bot.models import TelegramBot

from .models import *


class BaseTestCase(TestCase):
	def setUp(self) -> None:
		self.user: User = User.objects.create(telegram_id=123456789, first_name='exg1o')
		self.telegram_bot: TelegramBot = TelegramBot.objects.create(
			owner=self.user,
			api_token='123456789:qwertyuiop',
			is_private=True
		)
		self.plugin: Plugin = Plugin.objects.create(user=self.user, telegram_bot=self.telegram_bot, name='Test', code='def test():\n\tpass')
		self.plugin_log: PluginLog = PluginLog.objects.create(
			user=self.user,
			telegram_bot=self.telegram_bot,
			plugin=self.plugin,
			message='Error :-)',
			level='danger'
		)

class PluginModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.plugin.user, self.user)
		self.assertEqual(self.plugin.telegram_bot, self.telegram_bot)
		self.assertEqual(self.plugin.name, 'Test')
		self.assertEqual(self.plugin.code, 'def test():\n\tpass')
		self.assertFalse(self.plugin.is_checked)
		self.assertIsNotNone(self.plugin.added_date)

	def test_to_dict(self) -> None:
		self.assertDictEqual(self.plugin.to_dict(), {
			'id': 1,
			'name': 'Test',
			'code': 'def test():\n\tpass',
			'is_checked': False,
			'added_date': f'{filters.date(self.plugin.added_date)} {filters.time(self.plugin.added_date)}',
		})

class PluginLogModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.plugin_log.user, self.user)
		self.assertEqual(self.plugin_log.telegram_bot, self.telegram_bot)
		self.assertEqual(self.plugin_log.message, 'Error :-)')
		self.assertEqual(self.plugin_log.level, 'danger')
		self.assertIsNotNone(self.plugin_log.added_date)

	def test_to_dict(self) -> None:
		self.assertDictEqual(self.plugin_log.to_dict(), {
			'plugin_name': 'Test',
			'message': 'Error :-)',
			'level': 'danger',
			'added_date': f'{filters.date(self.plugin_log.added_date)} {filters.time(self.plugin_log.added_date)}',
		})

class PluginsViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('plugins', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('plugins', kwargs={'telegram_bot_id': 0})

	def test_post_method(self) -> None:
		response: HttpResponse = self.client.post(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(
			self.url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': '',
				'code': 'def test():\n\tpass',
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Тест',
				'code': 'def test():\n\tpass',
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test1',
				'code': 'def test():\n\tpass',
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(
			self.url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, [self.plugin.to_dict()])

class PluginViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('plugin', kwargs={'plugin_id': self.plugin.id})
		self.url_2: str = urls.reverse('plugin', kwargs={'plugin_id': 0})

	def test_patch_method(self) -> None:
		response: HttpResponse = self.client.patch(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.patch(
			self.url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'code': 'def test_():\n\tpass',
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_delete_method(self) -> None:
		response: HttpResponse = self.client.delete(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.delete(
			self.url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 200)

class ViewsTests(BaseTestCase):
	def test_get_plugins_logs_view(self) -> None:
		url_1: str = urls.reverse('plugins_logs', kwargs={'telegram_bot_id': self.telegram_bot.id})
		url_2: str = urls.reverse('plugins_logs', kwargs={'telegram_bot_id': 0})

		response: HttpResponse = self.client.get(url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(
			url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(
			url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, [self.plugin_log.to_dict()])

	def test_add_plugin_log_view(self) -> None:
		url_1: str = urls.reverse('plugin_logs', kwargs={'plugin_id': self.plugin.id})
		url_2: str = urls.reverse('plugin_logs', kwargs={'plugin_id': 0})

		response: HttpResponse = self.client.post(url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(
			url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(
			url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'message': ':D',
				'level': 'success',
			}
		)
		self.assertEqual(response.status_code, 200)

constructor_telegram_bots/plugin/tests.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('_/<int:telegram_bot_id>/', views.PluginsView.as_view(), name='plugins'),
    path('<int:plugin_id>/', views.PluginView.as_view(), name='plugin'),
    path('_/<int:telegram_bot_id>/logs/', views.get_plugins_logs_view, name='plugins_logs'),
	path('<int:plugin_id>/logs/', views.add_plugin_log_view, name='plugin_logs'),
]

constructor_telegram_bots/plugin/urls.py
pyguru-dev/telegrambot
from django.utils.translation import gettext as _

from rest_framework.decorators import APIView, api_view, authentication_classes, permission_classes
from rest_framework.authentication import TokenAuthentication
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response

from telegram_bot.models import TelegramBot
from telegram_bot.decorators import check_telegram_bot_id

from .models import *
from .decorators import *
from .serializers import *

from constructor_telegram_bots import environment


class PluginsView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	def post(self, request: Request, telegram_bot: TelegramBot) -> Response:
		serializer = CreatePluginSerializer(data=request.data, context={'telegram_bot': telegram_bot})
		serializer.is_valid(raise_exception=True)

		plugin: Plugin = Plugin.objects.create(
			user=request.user,
			telegram_bot=telegram_bot,
			**serializer.validated_data
		)

		return Response({
			'message': _('Вы успешно добавили плагин вашему Telegram боту.'),
			'level': 'success',

			'plugin': plugin.to_dict(),
		})

	@check_telegram_bot_id
	def get(self, request: Request, telegram_bot: TelegramBot) -> Response:
		return Response([plugin.to_dict() for plugin in telegram_bot.plugins.all()])

class PluginView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_plugin_id
	def patch(self, request: Request, plugin: Plugin) -> Response:
		serializer = UpdatePluginSerializer(data=request.data)
		serializer.is_valid(raise_exception=True)

		plugin.is_checked = False
		plugin.code = serializer.validated_data['code']
		plugin.save()

		environment.delete_plugin(plugin)

		return Response({
			'message': _('Вы успешно обновили плагин вашего Telegram бота.'),
			'level': 'success',

			'plugin': plugin.to_dict(),
		})

	@check_plugin_id
	def delete(self, request: Request, plugin: Plugin) -> Response:
		plugin.delete()

		return Response({
			'message': _('Вы успешно удалили плагин вашего Telegram бота.'),
			'level': 'success',
		})

@api_view(['GET'])
@authentication_classes([TokenAuthentication])
@permission_classes([IsAuthenticated])
@check_telegram_bot_id
def get_plugins_logs_view(request: Request, telegram_bot: TelegramBot) -> Response:
	return Response([plugin_log.to_dict() for plugin in telegram_bot.plugins.all() for plugin_log in plugin.logs.all()])

@api_view(['POST'])
@authentication_classes([TokenAuthentication])
@permission_classes([IsAuthenticated])
@check_plugin_id
def add_plugin_log_view(request: Request, plugin: Plugin) -> Response:
	serializer = AddPluginLogSerializer(data=request.data)
	serializer.is_valid(raise_exception=True)

	PluginLog.objects.create(
		user=request.user,
		telegram_bot=plugin.telegram_bot,
		plugin=plugin,
		**serializer.validated_data
	)

	return Response({
		'message': None,
		'level': 'success',
	})

constructor_telegram_bots/plugin/views.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls


class PrivacyPolicyViewsTests(TestCase):
	def test_privacy_policy_view(self) -> None:
		url: str = urls.reverse('privacy_policy')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'privacy_policy.html')

constructor_telegram_bots/privacy_policy/tests.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.privacy_policy_view, name='privacy_policy'),
]

constructor_telegram_bots/privacy_policy/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render


def privacy_policy_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'privacy_policy.html')

constructor_telegram_bots/privacy_policy/views.py
pyguru-dev/telegrambot
from django.contrib import admin
from modeltranslation.admin import TranslationAdmin

from .models import TeamMember


@admin.register(TeamMember)
class TeamMemberAdmin(TranslationAdmin):
	date_hierarchy = 'joined_date'
	list_filter = ('speciality', 'joined_date')

	list_display = ('username_', 'speciality', 'joined_date')
	fields = ('image', 'username', 'speciality', 'joined_date')

	@admin.display(description='@username', ordering='username')
	def username_(self, team_member: TeamMember) -> str:
		return f'@{team_member.username}'

constructor_telegram_bots/team/admin.py
pyguru-dev/telegrambot
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _


class TeamConfig(AppConfig):
	default_auto_field = 'django.db.models.BigAutoField'

	name = 'team'
	verbose_name = _('Команда')

constructor_telegram_bots/team/apps.py
pyguru-dev/telegrambot
from .models import TeamMember


def team_members(*args, **kwargs) -> dict:
    return {'team_members': TeamMember.objects.all()}

constructor_telegram_bots/team/context_processors.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-27 20:28

from django.db import migrations, models


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='TeamMembers',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('image', models.ImageField(upload_to='static/images/team/')),
                ('username', models.CharField(max_length=32, verbose_name='@username')),
                ('speciality', models.CharField(max_length=255, verbose_name='Специальность')),
                ('speciality_en', models.CharField(max_length=255, null=True, verbose_name='Специальность')),
                ('speciality_uk', models.CharField(max_length=255, null=True, verbose_name='Специальность')),
                ('speciality_ru', models.CharField(max_length=255, null=True, verbose_name='Специальность')),
                ('date_joined', models.DateTimeField(verbose_name='Дата присоединения')),
            ],
            options={
                'verbose_name': 'Члена команды',
                'verbose_name_plural': 'Члены команды',
                'db_table': 'team_members',
            },
        ),
    ]

constructor_telegram_bots/team/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-29 02:19

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('team', '0001_initial'),
    ]

    operations = [
        migrations.RenameModel(
            old_name='TeamMembers',
            new_name='TeamMember',
        ),
        migrations.AlterModelTable(
            name='teammember',
            table='team_member',
        ),
    ]

constructor_telegram_bots/team/migrations/0002_rename_teammembers_teammember_alter_teammember_table.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 21:34

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('team', '0002_rename_teammembers_teammember_alter_teammember_table'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='teammember',
            options={'verbose_name': 'Члена', 'verbose_name_plural': 'Члены'},
        ),
        migrations.RenameField(
            model_name='teammember',
            old_name='date_joined',
            new_name='joined_date',
        ),
    ]

constructor_telegram_bots/team/migrations/0003_alter_teammember_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 21:35

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('team', '0003_alter_teammember_options_and_more'),
    ]

    operations = [
        migrations.AlterField(
            model_name='teammember',
            name='joined_date',
            field=models.DateTimeField(verbose_name='Присоединился'),
        ),
    ]

constructor_telegram_bots/team/migrations/0004_alter_teammember_joined_date.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-20 02:13

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('team', '0004_alter_teammember_joined_date'),
    ]

    operations = [
        migrations.AlterField(
            model_name='teammember',
            name='image',
            field=models.ImageField(upload_to='team/'),
        ),
    ]

constructor_telegram_bots/team/migrations/0005_alter_teammember_image.py
pyguru-dev/telegrambot
from django.db import models
from django.utils.translation import gettext_lazy as _


class TeamMember(models.Model):
	image = models.ImageField(upload_to='team/')
	username = models.CharField('@username', max_length=32)
	speciality = models.CharField(_('Специальность'), max_length=255)
	joined_date = models.DateTimeField(_('Присоединился'))

	class Meta:
		db_table = 'team_member'

		verbose_name = _('Члена')
		verbose_name_plural = _('Члены')

	def delete(self) -> None:
		self.image.delete(save=False)
		return super().delete()

	def __str__(self) -> str:
		return f'@{self.username}'

constructor_telegram_bots/team/models.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls

from .models import TeamMember

from datetime import datetime


class TeamMemberModelTests(TestCase):
	def setUp(self) -> None:
		self.current_date: datetime = datetime.now()
		self.team_member: TeamMember = TeamMember.objects.create(
			username='test',
			speciality='Test',
			joined_date=self.current_date
		)

	def test_fields(self) -> None:
		self.assertEqual(str(self.team_member.image), '')
		self.assertEqual(self.team_member.username, 'test')
		self.assertEqual(self.team_member.speciality, 'Test')
		self.assertEqual(self.team_member.joined_date, self.current_date)

class ViewsTests(TestCase):
	def test_team_view(self) -> None:
		url: str = urls.reverse('team')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'team.html')

constructor_telegram_bots/team/tests.py
pyguru-dev/telegrambot
from modeltranslation.translator import register, TranslationOptions

from .models import TeamMember


@register(TeamMember)
class TeamMemberTranslationOptions(TranslationOptions):
    fields = ('speciality',)

constructor_telegram_bots/team/translation.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.team_view, name='team'),
]

constructor_telegram_bots/team/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render


def team_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'team.html')

constructor_telegram_bots/team/views.py
pyguru-dev/telegrambot
from django.contrib import admin, messages
from django.utils import html
from django.http import HttpRequest
from django.utils.translation import gettext_lazy as _

from .models import *
from .services import tasks

from typing import List
import sys


@admin.register(TelegramBot)
class TelegramBotAdmin(admin.ModelAdmin):
	search_fields = ('username',)
	date_hierarchy = 'added_date'
	list_filter = ('is_running', 'added_date')

	actions = ('start_telegram_bot_action', 'stop_telegram_bot_action')

	list_display = ('id', 'owner', 'username_', 'is_private', 'is_running', 'commands_count', 'users_count', 'added_date')

	fields = ('id', 'owner', 'username_', 'api_token', 'is_private', 'is_running', 'commands_count', 'users_count', 'added_date')

	@admin.display(description='@username', ordering='username')
	def username_(self, telegram_bot: TelegramBot) -> str:
		return html.format_html(f'<a href="tg://resolve?domain={telegram_bot.username}" style="font-weight: 600;" target="_blank">@{telegram_bot.username}</a>')

	@admin.display(description=_('Команд'))
	def commands_count(self, telegram_bot: TelegramBot) -> int:
		return telegram_bot.commands.count()

	@admin.display(description=_('Активаций'))
	def users_count(self, telegram_bot: TelegramBot) -> int:
		return telegram_bot.users.count()

	@admin.action(description=_('Включить Telegram бота'))
	def start_telegram_bot_action(self, request: HttpRequest, telegram_bots: List[TelegramBot]) -> None:
		for telegram_bot in telegram_bots:
			if not telegram_bot.is_running and telegram_bot.is_stopped:
				if sys.platform == 'win32':
					tasks.start_telegram_bot(telegram_bot_id=telegram_bot.id)
				else:
					tasks.start_telegram_bot.delay(telegram_bot_id=telegram_bot.id)

				messages.success(request, f"@{telegram_bot.username} {_('Telegram бот успешно включен.')}")
			else:
				messages.error(request, f"@{telegram_bot.username} {_('Telegram бот уже включен!')}")

	@admin.action(description=_('Выключить Telegram бота'))
	def stop_telegram_bot_action(self, request: HttpRequest, telegram_bots: List[TelegramBot]) -> None:
		for telegram_bot in telegram_bots:
			if telegram_bot.is_running and not telegram_bot.is_stopped:
				if sys.platform == 'win32':
					tasks.stop_telegram_bot(telegram_bot_id=telegram_bot.id)
				else:
					tasks.stop_telegram_bot.delay(telegram_bot_id=telegram_bot.id)

				messages.success(request, f"@{telegram_bot.username} {_('Telegram бот успешно выключен.')}")
			else:
				messages.error(request, f"@{telegram_bot.username} {_('Telegram бот уже выключен!')}")

	def has_add_permission(self, *args, **kwargs) -> bool:
		return False

	def has_change_permission(self, *args, **kwargs) -> bool:
		return False

@admin.register(TelegramBotUser)
class TelegramBotUserAdmin(admin.ModelAdmin):
	search_fields = ('user_id', 'full_name')
	date_hierarchy = 'activated_date'
	list_filter = ('activated_date',)

	list_display = ('id', 'telegram_bot', 'user_id', 'full_name', 'is_allowed', 'activated_date')

	fields = ('id', 'telegram_bot', 'user_id', 'full_name', 'is_allowed', 'activated_date')

	def has_add_permission(self, *args, **kwargs) -> bool:
		return False

	def has_change_permission(self, *args, **kwargs) -> bool:
		return False

constructor_telegram_bots/telegram_bot/admin.py
pyguru-dev/telegrambot
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _


class TelegramBotConfig(AppConfig):
	default_auto_field = 'django.db.models.BigAutoField'

	name = 'telegram_bot'
	verbose_name = _('Telegram боты')

constructor_telegram_bots/telegram_bot/apps.py
pyguru-dev/telegrambot
from .models import TelegramBot


def telegram_bots(*args, **kwargs) -> dict:
    return {'telegram_bots': TelegramBot.objects.all()}

constructor_telegram_bots/telegram_bot/context_processors.py
pyguru-dev/telegrambot
from django.http import HttpRequest, JsonResponse
from django.utils.translation import gettext as _

from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.status import *

from .models import TelegramBot, TelegramBotCommand, TelegramBotCommandKeyboard

from functools import wraps
from typing import Union


def check_telegram_bot_id(func):
	@wraps(func)
	def wrapper(*args, **kwargs):
		request: Union[HttpRequest, Request] = args[-1]
		telegram_bot_id: int = kwargs.pop('telegram_bot_id')

		if not request.user.telegram_bots.filter(id=telegram_bot_id).exists():
			return (JsonResponse if isinstance(request, HttpRequest) else Response)({
				'message': _('Telegram бот не найден!'),
				'level': 'danger',
			}, status=HTTP_404_NOT_FOUND)

		return func(telegram_bot=request.user.telegram_bots.get(id=telegram_bot_id), *args, **kwargs)
	return wrapper

def check_telegram_bot_command_id(func):
	@wraps(func)
	def wrapper(*args, **kwargs):
		telegram_bot: TelegramBot = kwargs['telegram_bot']
		telegram_bot_command_id: int = kwargs.pop('telegram_bot_command_id')

		if not telegram_bot.commands.filter(id=telegram_bot_command_id).exists():
			return Response({
				'message': _('Команда Telegram бота не найдена!'),
				'level': 'danger',
			}, status=HTTP_404_NOT_FOUND)

		return func(telegram_bot_command=telegram_bot.commands.get(id=telegram_bot_command_id), *args, **kwargs)
	return wrapper

def check_telegram_bot_command_keyboard_button_id(func):
	@wraps(func)
	def wrapper(*args, **kwargs):
		telegram_bot_command: TelegramBotCommand = kwargs['telegram_bot_command']
		telegram_bot_command_keyboard: TelegramBotCommandKeyboard = telegram_bot_command.keyboard
		telegram_bot_command_keyboard_button_id: int = kwargs.pop('telegram_bot_command_keyboard_button_id')

		if not telegram_bot_command_keyboard.buttons.filter(id=telegram_bot_command_keyboard_button_id).exists():
			return Response({
				'message': _('Кнопка клавиатуры команды Telegram бота не найдена!'),
				'level': 'danger',
			}, status=HTTP_404_NOT_FOUND)

		return func(telegram_bot_command_keyboard_button=telegram_bot_command_keyboard.buttons.get(id=telegram_bot_command_keyboard_button_id), *args, **kwargs)
	return wrapper

def check_telegram_bot_user_id(func):
	def wrapper(*args, **kwargs):
		telegram_bot: TelegramBot = kwargs['telegram_bot']
		telegram_bot_user_id: int = kwargs.pop('telegram_bot_user_id')

		if not telegram_bot.users.filter(id=telegram_bot_user_id).exists():
			return Response({
				'message': _('Пользователь Telegram бота не найдена!'),
				'level': 'danger',
			}, status=HTTP_404_NOT_FOUND)

		return func(telegram_bot_user=telegram_bot.users.get(id=telegram_bot_user_id), *args, **kwargs)
	return wrapper

constructor_telegram_bots/telegram_bot/decorators.py
pyguru-dev/telegrambot
from django.conf import settings
from django.core.files.uploadedfile import InMemoryUploadedFile

from rest_framework.request import Request

from typing import Optional, Union
import requests


def check_telegram_bot_api_token(api_token: str) -> Optional[str]:
	if settings.TEST:
		return f'{api_token}_test_telegram_bot'

	responce: requests.Response = requests.get(f'https://api.telegram.org/bot{api_token}/getMe')

	if responce.status_code == 200:
		return responce.json()['result']['username']

def get_image_from_request(request: Request) -> Union[InMemoryUploadedFile, str, None]:
	if 'image' in request.FILES:
		return request.FILES['image']
	elif 'image' in request.POST:
		return request.POST['image']

	return None

constructor_telegram_bots/telegram_bot/functions.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-05-17 15:42

from django.db import migrations, models


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='TelegramBotCommand',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('name', models.CharField(max_length=255)),
                ('command', models.CharField(max_length=32, null=True)),
                ('callback', models.CharField(max_length=64, null=True)),
                ('message_text', models.TextField()),
                ('keyboard', models.JSONField()),
            ],
            options={
                'db_table': 'telegram_bot_command',
            },
        ),
        migrations.CreateModel(
            name='TelegramBotUser',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('user_id', models.BigIntegerField()),
                ('username', models.CharField(max_length=32)),
                ('is_allowed', models.BooleanField(default=False)),
                ('date_activated', models.DateTimeField(auto_now_add=True)),
            ],
            options={
                'db_table': 'telegram_bot_user',
            },
        ),
        migrations.CreateModel(
            name='TelegramBot',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('name', models.CharField(max_length=32, unique=True)),
                ('api_token', models.CharField(max_length=50, unique=True)),
                ('is_private', models.BooleanField(default=True)),
                ('is_running', models.BooleanField(default=False)),
                ('is_stopped', models.BooleanField(default=True)),
                ('date_added', models.DateTimeField(auto_now_add=True)),
                ('commands', models.ManyToManyField(related_name='commands', to='telegram_bot.telegrambotcommand')),
                ('users', models.ManyToManyField(related_name='users', to='telegram_bot.telegrambotuser')),
            ],
            options={
                'db_table': 'telegram_bot',
            },
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-01 21:44

from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
        ('telegram_bot', '0001_initial'),
    ]

    operations = [
        migrations.CreateModel(
            name='TelegramBotCommandKeyboard',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('type', models.CharField(choices=[('default', 'Default'), ('inline', 'Inline')], default='default', max_length=7)),
            ],
            options={
                'db_table': 'telegram_bot_command_keyboard',
            },
        ),
        migrations.RenameField(
            model_name='telegrambot',
            old_name='date_added',
            new_name='_date_added',
        ),
        migrations.RenameField(
            model_name='telegrambotuser',
            old_name='date_activated',
            new_name='_date_activated',
        ),
        migrations.RemoveField(
            model_name='telegrambot',
            name='commands',
        ),
        migrations.RemoveField(
            model_name='telegrambot',
            name='users',
        ),
        migrations.RemoveField(
            model_name='telegrambotcommand',
            name='callback',
        ),
        migrations.RemoveField(
            model_name='telegrambotcommand',
            name='keyboard',
        ),
        migrations.AddField(
            model_name='telegrambot',
            name='diagram_current_scale',
            field=models.FloatField(default=1.0),
        ),
        migrations.AddField(
            model_name='telegrambot',
            name='owner',
            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='telegram_bots', to=settings.AUTH_USER_MODEL),
        ),
        migrations.AddField(
            model_name='telegrambotcommand',
            name='api_request',
            field=models.JSONField(null=True),
        ),
        migrations.AddField(
            model_name='telegrambotcommand',
            name='image',
            field=models.ImageField(null=True, upload_to='static/images/'),
        ),
        migrations.AddField(
            model_name='telegrambotcommand',
            name='telegram_bot',
            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='commands', to='telegram_bot.telegrambot'),
        ),
        migrations.AddField(
            model_name='telegrambotcommand',
            name='x',
            field=models.IntegerField(default=0),
        ),
        migrations.AddField(
            model_name='telegrambotcommand',
            name='y',
            field=models.IntegerField(default=0),
        ),
        migrations.AddField(
            model_name='telegrambotuser',
            name='telegram_bot',
            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='users', to='telegram_bot.telegrambot'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='is_private',
            field=models.BooleanField(),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='message_text',
            field=models.TextField(max_length=4096),
        ),
        migrations.CreateModel(
            name='TelegramBotCommandKeyboardButton',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('text', models.TextField(max_length=4096)),
                ('start_diagram_connector', models.TextField(null=True)),
                ('end_diagram_connector', models.TextField(null=True)),
                ('telegram_bot_command', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, to='telegram_bot.telegrambotcommand')),
                ('telegram_bot_command_keyboard', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='buttons', to='telegram_bot.telegrambotcommandkeyboard')),
            ],
            options={
                'db_table': 'telegram_bot_command_keyboard_button',
            },
        ),
        migrations.AddField(
            model_name='telegrambotcommandkeyboard',
            name='telegram_bot_command',
            field=models.OneToOneField(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='keyboard', to='telegram_bot.telegrambotcommand'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0002_telegrambotcommandkeyboard_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-18 07:48

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0002_telegrambotcommandkeyboard_and_more'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='telegrambotuser',
            name='username',
        ),
        migrations.AddField(
            model_name='telegrambotuser',
            name='full_name',
            field=models.CharField(max_length=129, null=True),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0003_remove_telegrambotuser_username_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-21 16:31

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0003_remove_telegrambotuser_username_and_more'),
    ]

    operations = [
        migrations.AddField(
            model_name='telegrambotcommandkeyboardbutton',
            name='url',
            field=models.TextField(max_length=2048, null=True),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0004_telegrambotcommandkeyboardbutton_url.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-23 00:47

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0004_telegrambotcommandkeyboardbutton_url'),
    ]

    operations = [
        migrations.AddField(
            model_name='telegrambotcommandkeyboardbutton',
            name='row',
            field=models.IntegerField(null=True),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0005_telegrambotcommandkeyboardbutton_row.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-24 13:18

from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
        ('telegram_bot', '0005_telegrambotcommandkeyboardbutton_row'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='telegrambot',
            options={'verbose_name': 'Telegram бота', 'verbose_name_plural': 'Telegram боты'},
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='_date_added',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Дата добавления'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='is_private',
            field=models.BooleanField(verbose_name='Приватный'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='is_running',
            field=models.BooleanField(default=False, verbose_name='Запущен'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='name',
            field=models.CharField(max_length=32, unique=True, verbose_name='@username'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='owner',
            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='telegram_bots', to=settings.AUTH_USER_MODEL, verbose_name='Владелец'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0006_alter_telegrambot_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-25 19:12

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0006_alter_telegrambot_options_and_more'),
    ]

    operations = [
        migrations.RenameField(
            model_name='telegrambot',
            old_name='name',
            new_name='username',
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0007_rename_name_telegrambot_username.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-27 08:09

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0007_rename_name_telegrambot_username'),
    ]

    operations = [
        migrations.AlterField(
            model_name='telegrambot',
            name='is_running',
            field=models.BooleanField(default=False, verbose_name='Включён'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0008_alter_telegrambot_is_running.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-27 08:49

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0008_alter_telegrambot_is_running'),
    ]

    operations = [
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='image',
            field=models.ImageField(null=True, upload_to='static/images/commands/'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0009_alter_telegrambotcommand_image.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-01 22:41

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0009_alter_telegrambotcommand_image'),
    ]

    operations = [
        migrations.RenameField(
            model_name='telegrambot',
            old_name='_date_added',
            new_name='date_added',
        ),
        migrations.RenameField(
            model_name='telegrambotuser',
            old_name='_date_activated',
            new_name='date_activated',
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0010_rename__date_added_telegrambot_date_added_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-02 03:39

from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0010_rename__date_added_telegrambot_date_added_and_more'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='telegrambotcommand',
            options={'verbose_name': 'Команда Telegram бота', 'verbose_name_plural': 'Команды Telegram ботов'},
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='api_request',
            field=models.JSONField(blank=True, null=True, verbose_name='API-запрос'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='command',
            field=models.CharField(blank=True, max_length=32, null=True, verbose_name='Команда'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='image',
            field=models.ImageField(blank=True, null=True, upload_to='static/images/commands/'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='message_text',
            field=models.TextField(max_length=4096, verbose_name='Текст сообщения'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='name',
            field=models.CharField(max_length=255, verbose_name='Название'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='telegram_bot',
            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='commands', to='telegram_bot.telegrambot', verbose_name='Telegram бот'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='x',
            field=models.IntegerField(default=0, verbose_name='Координата X'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='y',
            field=models.IntegerField(default=0, verbose_name='Координата Y'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0011_alter_telegrambotcommand_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-06 03:37

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0011_alter_telegrambotcommand_options_and_more'),
    ]

    operations = [
        migrations.AddField(
            model_name='telegrambotcommand',
            name='database_record',
            field=models.JSONField(blank=True, null=True, verbose_name='Запись в базу данных'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0012_telegrambotcommand_database_record.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-06 04:34

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0012_telegrambotcommand_database_record'),
    ]

    operations = [
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='database_record',
            field=models.TextField(blank=True, null=True, verbose_name='Запись в базу данных'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0013_alter_telegrambotcommand_database_record.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-11 21:24

from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0013_alter_telegrambotcommand_database_record'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='telegrambotuser',
            options={'verbose_name': 'Пользователя Telegram бота', 'verbose_name_plural': 'Пользователи Telegram ботов'},
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='date_activated',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Дата активации'),
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='full_name',
            field=models.CharField(max_length=129, null=True, verbose_name='Полное имя пользователя'),
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='is_allowed',
            field=models.BooleanField(default=False, verbose_name='Разрешён'),
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='telegram_bot',
            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='users', to='telegram_bot.telegrambot', verbose_name='Telegram бот'),
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='user_id',
            field=models.BigIntegerField(verbose_name='Telegram ID пользователя'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0014_alter_telegrambotuser_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-08-02 17:04

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0014_alter_telegrambotuser_options_and_more'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='telegrambotcommandkeyboardbutton',
            options={'ordering': ['id']},
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0015_alter_telegrambotcommandkeyboardbutton_options.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-11 04:47

from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0015_alter_telegrambotcommandkeyboardbutton_options'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='telegrambotcommand',
            options={'verbose_name': 'Команда', 'verbose_name_plural': 'Команды'},
        ),
        migrations.AlterModelOptions(
            name='telegrambotuser',
            options={'verbose_name': 'Пользователя', 'verbose_name_plural': 'Пользователи'},
        ),
        migrations.RenameField(
            model_name='telegrambot',
            old_name='date_added',
            new_name='added_date',
        ),
        migrations.RenameField(
            model_name='telegrambotuser',
            old_name='date_activated',
            new_name='activated_date',
        ),
        migrations.RemoveField(
            model_name='telegrambotcommand',
            name='api_request',
        ),
        migrations.RemoveField(
            model_name='telegrambotcommand',
            name='command',
        ),
        migrations.RemoveField(
            model_name='telegrambotcommand',
            name='message_text',
        ),
        migrations.RemoveField(
            model_name='telegrambotcommandkeyboard',
            name='type',
        ),
        migrations.AddField(
            model_name='telegrambotcommandkeyboard',
            name='mode',
            field=models.CharField(choices=[('default', 'Обычный'), ('inline', 'Встроенный'), ('payment', 'Платёжный')], default='default', max_length=7, verbose_name='Режим'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='api_token',
            field=models.CharField(max_length=50, unique=True, verbose_name='API-токен'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='database_record',
            field=models.JSONField(blank=True, null=True, verbose_name='Запись в базу данных'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='image',
            field=models.ImageField(blank=True, null=True, upload_to='static/images/commands/', verbose_name='Изображение'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandkeyboardbutton',
            name='end_diagram_connector',
            field=models.TextField(blank=True, null=True),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandkeyboardbutton',
            name='row',
            field=models.IntegerField(blank=True, null=True, verbose_name='Ряд'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandkeyboardbutton',
            name='start_diagram_connector',
            field=models.TextField(blank=True, null=True),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandkeyboardbutton',
            name='telegram_bot_command',
            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='telegram_bot.telegrambotcommand'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandkeyboardbutton',
            name='text',
            field=models.TextField(max_length=4096, verbose_name='Текст'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandkeyboardbutton',
            name='url',
            field=models.TextField(blank=True, max_length=2048, null=True, verbose_name='URL-адрес'),
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='full_name',
            field=models.CharField(max_length=129, null=True, verbose_name='Полное имя'),
        ),
        migrations.AlterField(
            model_name='telegrambotuser',
            name='user_id',
            field=models.BigIntegerField(verbose_name='Telegram ID'),
        ),
        migrations.CreateModel(
            name='TelegramBotCommandMessageText',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('mode', models.CharField(choices=[('default', 'Обычный'), ('markdown', 'Markdown'), ('html', 'HTML')], default='default', max_length=8, verbose_name='Режим')),
                ('text', models.TextField(max_length=4096, verbose_name='Текст')),
                ('telegram_bot_command', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='message_text', to='telegram_bot.telegrambotcommand')),
            ],
            options={
                'db_table': 'telegram_bot_command_message_text',
            },
        ),
        migrations.CreateModel(
            name='TelegramBotCommandCommand',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('command', models.CharField(max_length=32, verbose_name='Команда')),
                ('show_in_menu', models.BooleanField(default=False, verbose_name='Отображать в меню')),
                ('telegram_bot_command', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='command', to='telegram_bot.telegrambotcommand')),
            ],
            options={
                'db_table': 'telegram_bot_command_command',
            },
        ),
        migrations.CreateModel(
            name='TelegramBotCommandApiRequest',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('url', models.URLField(verbose_name='URL-адрес')),
                ('method', models.CharField(choices=[('get', 'GET'), ('post', 'POST'), ('put', 'PUT'), ('patch', 'PATCH'), ('delete', 'DELETE')], default='get', max_length=6, verbose_name='Метод')),
                ('headers', models.JSONField(blank=True, null=True, verbose_name='Заголовки')),
                ('data', models.JSONField(blank=True, null=True, verbose_name='Данные')),
                ('telegram_bot_command', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='api_request', to='telegram_bot.telegrambotcommand')),
            ],
            options={
                'db_table': 'telegram_bot_command_api_request',
            },
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0016_alter_telegrambotcommand_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-20 01:31

from django.db import migrations, models
import telegram_bot.models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0016_alter_telegrambotcommand_options_and_more'),
    ]

    operations = [
        migrations.AlterField(
            model_name='telegrambotcommand',
            name='image',
            field=models.ImageField(blank=True, null=True, upload_to=telegram_bot.models.upload_telegram_bot_command_image_path, verbose_name='Изображение'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0017_alter_telegrambotcommand_image.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-22 20:50

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0017_alter_telegrambotcommand_image'),
    ]

    operations = [
        migrations.RenameField(
            model_name='telegrambotcommandcommand',
            old_name='show_in_menu',
            new_name='is_show_in_menu',
        ),
        migrations.RenameField(
            model_name='telegrambotcommandcommand',
            old_name='command',
            new_name='text',
        ),
        migrations.AddField(
            model_name='telegrambotcommandcommand',
            name='description',
            field=models.CharField(blank=True, max_length=255, null=True, verbose_name='Описание'),
        ),
        migrations.AlterField(
            model_name='telegrambot',
            name='is_private',
            field=models.BooleanField(default=False, verbose_name='Приватный'),
        ),
        migrations.AlterField(
            model_name='telegrambotcommandapirequest',
            name='url',
            field=models.TextField(max_length=2048, verbose_name='URL-адрес'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0018_rename_show_in_menu_telegrambotcommandcommand_is_show_in_menu_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-24 00:02

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0018_rename_show_in_menu_telegrambotcommandcommand_is_show_in_menu_and_more'),
    ]

    operations = [
        migrations.AlterField(
            model_name='telegrambot',
            name='added_date',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Добавлен'),
        ),
    ]

constructor_telegram_bots/telegram_bot/migrations/0019_alter_telegrambot_added_date.py
pyguru-dev/telegrambot
from django.db import models
from django.utils.translation import gettext_lazy as _
from django.core.exceptions import ObjectDoesNotExist
from django.template import defaultfilters as filters
from django.core.files.uploadedfile import InMemoryUploadedFile

from user.models import User

from .services import database_telegram_bot
from .functions import check_telegram_bot_api_token

from asgiref.sync import sync_to_async
from typing import Optional, Union


class TelegramBotManager(models.Manager):
	def create(self, owner: User, api_token: str, is_private: bool, **extra_fields) -> 'TelegramBot':
		username: str = check_telegram_bot_api_token(api_token)

		return super().create(
			owner=owner,
			username=username,
			api_token=api_token,
			is_private=is_private,
			**extra_fields
		)

class TelegramBot(models.Model):
	owner = models.ForeignKey(User, on_delete=models.CASCADE, related_name='telegram_bots', verbose_name=_('Владелец'), null=True)
	username = models.CharField('@username', max_length=32, unique=True)
	api_token = models.CharField(_('API-токен'), max_length=50, unique=True)
	is_private = models.BooleanField(_('Приватный'), default=False)
	is_running = models.BooleanField(_('Включён'), default=False)
	is_stopped = models.BooleanField(default=True)
	added_date = models.DateTimeField(_('Добавлен'), auto_now_add=True)

	diagram_current_scale = models.FloatField(default=1.0)

	objects = TelegramBotManager()

	class Meta:
		db_table = 'telegram_bot'

		verbose_name = _('Telegram бота')
		verbose_name_plural = _('Telegram боты')

	def get_commands_as_dict(self, escape: bool = False) -> list:
		return [command.to_dict(escape=escape) for command in self.commands.all()]

	def get_users_as_dict(self) -> list:
		return [user.to_dict() for user in self.users.all()]

	def to_dict(self) -> dict:
		return {
			'id': self.id,
			'username': self.username,
			'api_token': self.api_token,
			'is_private': self.is_private,
			'is_running': self.is_running,
			'is_stopped': self.is_stopped,
			'commands_count': self.commands.count(),
			'users_count': self.users.count(),
			'added_date': f'{filters.date(self.added_date)} {filters.time(self.added_date)}',
		}

	def delete(self) -> None:
		database_telegram_bot.delete_collection(self)
		super().delete()

	def __str__(self) -> str:
		return f'@{self.username}'

class TelegramBotCommandManager(models.Manager):
	def create(
		self,
		telegram_bot: TelegramBot,
		name: str,
		command: Optional[dict],
		image: Union[InMemoryUploadedFile, str, None],
		message_text: dict,
		keyboard: Optional[dict],
		api_request: Optional[dict],
		database_record: Optional[dict],
	) -> 'TelegramBotCommand':
		telegram_bot_command: TelegramBotCommand = super().create(
			telegram_bot=telegram_bot,
			name=name,
			image=image if isinstance(image, InMemoryUploadedFile) else None,
			database_record=database_record
		)

		if command:
			TelegramBotCommandCommand.objects.create(telegram_bot_command=telegram_bot_command, **command)

		TelegramBotCommandMessageText.objects.create(telegram_bot_command=telegram_bot_command, **message_text)

		if keyboard:
			TelegramBotCommandKeyboard.objects.create(telegram_bot_command=telegram_bot_command, **keyboard)

		if api_request:
			TelegramBotCommandApiRequest.objects.create(telegram_bot_command=telegram_bot_command, **api_request)

		return telegram_bot_command

	async def acreate(self, *args, **kwargs) -> 'TelegramBotCommand':
		return await sync_to_async(self.create)(*args, **kwargs)

def upload_telegram_bot_command_image_path(telegram_bot_command: 'TelegramBotCommand', file_name: str):
	return f'telegram_bots/{telegram_bot_command.telegram_bot.id}/commands/{file_name}'

class TelegramBotCommand(models.Model):
	telegram_bot = models.ForeignKey(TelegramBot, on_delete=models.CASCADE, related_name='commands', null=True, verbose_name=_('Telegram бот'))
	name = models.CharField(_('Название'), max_length=255)
	image = models.ImageField(_('Изображение'), upload_to=upload_telegram_bot_command_image_path, blank=True, null=True)
	database_record = models.JSONField(_('Запись в базу данных'), blank=True, null=True)

	x =	models.IntegerField(_('Координата X'), default=0)
	y = models.IntegerField(_('Координата Y'), default=0)

	objects = TelegramBotCommandManager()

	class Meta:
		db_table = 'telegram_bot_command'

		verbose_name = _('Команда')
		verbose_name_plural = _('Команды')

	def get_command(self) -> Optional['TelegramBotCommandCommand']:
		try:
			return self.command
		except ObjectDoesNotExist:
			return None

	async def aget_command(self) -> Optional['TelegramBotCommandCommand']:
		return await sync_to_async(self.get_command)()

	def get_message_text(self) -> Optional['TelegramBotCommandMessageText']:
		try:
			return self.message_text
		except ObjectDoesNotExist:
			return None

	async def aget_message_text(self) -> Optional['TelegramBotCommandMessageText']:
		return await sync_to_async(self.get_message_text)()

	def get_keyboard(self) -> Optional['TelegramBotCommandKeyboard']:
		try:
			return self.keyboard
		except ObjectDoesNotExist:
			return None

	async def aget_keyboard(self) -> Optional['TelegramBotCommandKeyboard']:
		return await sync_to_async(self.get_keyboard)()

	def get_keyboard_as_dict(self, escape: bool = False) -> Optional[dict]:
		keyboard: Optional[TelegramBotCommandKeyboard] = self.get_keyboard()
		return keyboard.to_dict(escape=escape) if keyboard else None

	def get_api_request(self) -> Optional['TelegramBotCommandApiRequest']:
		try:
			return self.api_request
		except ObjectDoesNotExist:
			return None

	async def aget_api_request(self) -> Optional['TelegramBotCommandApiRequest']:
		return await sync_to_async(self.get_api_request)()

	def to_dict(self, escape: bool = False) -> dict:
		command: Optional[TelegramBotCommandCommand] = self.get_command()
		api_request: Optional[TelegramBotCommandApiRequest] = self.get_api_request()

		return {
			'id': self.id,
			'name': filters.escape(self.name) if escape else self.name,
			'command': command.to_dict() if command else None,
			'image': self.image.url if self.image else None,
			'message_text': self.message_text.to_dict(escape=escape),
			'keyboard': self.get_keyboard_as_dict(escape=escape),
			'api_request': api_request.to_dict() if api_request else None,
			'database_record': self.database_record,

			'x': self.x,
			'y': self.y,
		}

	def update(
		self,
		name: str,
		command: Optional[dict],
		image: Union[InMemoryUploadedFile, str, None],
		message_text: dict,
		keyboard: Optional[dict],
		api_request: Optional[dict],
		database_record: Optional[dict],
	):
		self.name = name
		self.database_record = database_record
		self.save()

		telegram_bot_command_command: Optional[TelegramBotCommandCommand] = self.get_command()

		if command:
			if telegram_bot_command_command:
				telegram_bot_command_command.text = command['text']
				telegram_bot_command_command.is_show_in_menu = command['is_show_in_menu']
				telegram_bot_command_command.description = command['description']
				telegram_bot_command_command.save()
			else:
				TelegramBotCommandCommand.objects.create(telegram_bot_command=self, **command)
		else:
			if telegram_bot_command_command:
				telegram_bot_command_command.delete()

		if isinstance(image, InMemoryUploadedFile) or image == 'null' and self.image != None:
			self.image.delete(save=False)

			if isinstance(image, InMemoryUploadedFile):
				self.image = image

			self.save()

		self.message_text.mode = message_text['mode']
		self.message_text.text = message_text['text']
		self.message_text.save()

		telegram_bot_command_keyboard: Optional[TelegramBotCommandKeyboard] = self.get_keyboard()

		if keyboard:
			if telegram_bot_command_keyboard:
				telegram_bot_command_keyboard.mode = keyboard['mode']
				telegram_bot_command_keyboard.save()

				buttons_id = []

				for button in keyboard['buttons']:
					is_finded_button = False

					if button['id']:
						button_id = int(button['id'])

						for button_ in telegram_bot_command_keyboard.buttons.all():
							if button_id == button_.id:
								is_finded_button = True
								break

					if is_finded_button:
						button_: TelegramBotCommandKeyboardButton = telegram_bot_command_keyboard.buttons.get(id=button_id)
						button_.row = button['row']
						button_.text = button['text']
						button_.url = button['url']
						button_.save()
					else:
						button_: TelegramBotCommandKeyboardButton = TelegramBotCommandKeyboardButton.objects.create(
							telegram_bot_command_keyboard=telegram_bot_command_keyboard,
							**button
						)

					buttons_id.append(button_.id)

				for button in telegram_bot_command_keyboard.buttons.all():
					if button.id not in buttons_id:
						button.delete()
			else:
				TelegramBotCommandKeyboard.objects.create(telegram_bot_command=self, **keyboard)
		else:
			if telegram_bot_command_keyboard:
				telegram_bot_command_keyboard.delete()

		telegram_bot_command_api_request: Optional[TelegramBotCommandApiRequest] = self.get_api_request()

		if api_request:
			if telegram_bot_command_api_request:
				telegram_bot_command_api_request.url = api_request['url']
				telegram_bot_command_api_request.method = api_request['method']
				telegram_bot_command_api_request.headers = api_request['headers']
				telegram_bot_command_api_request.data = api_request['data']
				telegram_bot_command_api_request.save()
			else:
				TelegramBotCommandApiRequest.objects.create(telegram_bot_command=self, **api_request)
		else:
			if telegram_bot_command_api_request:
				telegram_bot_command_api_request.delete()

	def delete(self) -> None:
		self.image.delete(save=False)
		return super().delete()

	def __str__(self) -> str:
		return self.name

class TelegramBotCommandCommand(models.Model):
	telegram_bot_command = models.OneToOneField(TelegramBotCommand, on_delete=models.CASCADE, related_name='command')
	text = models.CharField(_('Команда'), max_length=32)
	is_show_in_menu = models.BooleanField(_('Отображать в меню'), default=False)
	description = models.CharField(_('Описание'), max_length=255, blank=True, null=True)

	class Meta:
		db_table = 'telegram_bot_command_command'

	def to_dict(self) -> dict:
		return {
			'text': self.text,
			'is_show_in_menu': self.is_show_in_menu,
			'description': self.description,
		}

class TelegramBotCommandMessageText(models.Model):
	telegram_bot_command = models.OneToOneField(TelegramBotCommand, on_delete=models.CASCADE, related_name='message_text')
	mode = models.CharField(_('Режим'), max_length=8, choices=(
		('default', _('Обычный')),
		('markdown', 'Markdown'),
		('html', 'HTML'),
	), default='default')
	text = models.TextField(_('Текст'), max_length=4096)

	class Meta:
		db_table = 'telegram_bot_command_message_text'

	def to_dict(self, escape: bool = False) -> dict:
		return {
			'mode': self.mode,
			'text': filters.escape(self.text) if escape else self.text,
		}

class TelegramBotCommandKeyboardManager(models.Manager):
	def create(
		self,
		telegram_bot_command: TelegramBotCommand,
		mode: str,
		buttons: list
	) -> 'TelegramBotCommandKeyboard':
		telegram_bot_command_keyboard: TelegramBotCommandKeyboard = super().create(telegram_bot_command=telegram_bot_command, mode=mode)

		for button in buttons:
			TelegramBotCommandKeyboardButton.objects.create(
				telegram_bot_command_keyboard=telegram_bot_command_keyboard,
				row=button['row'],
				text=button['text'],
				url=button['url']
			)

		return telegram_bot_command_keyboard

class TelegramBotCommandKeyboard(models.Model):
	telegram_bot_command = models.OneToOneField(TelegramBotCommand, on_delete=models.CASCADE, related_name='keyboard', null=True)
	mode = models.CharField(_('Режим'), max_length=7, choices=(
		('default', _('Обычный')),
		('inline', _('Встроенный')),
		('payment', _('Платёжный')),
   	), default='default')

	objects = TelegramBotCommandKeyboardManager()

	class Meta:
		db_table = 'telegram_bot_command_keyboard'

	def get_buttons_as_dict(self, escape: bool = False) -> list:
		return [button.to_dict(escape=escape) for button in self.buttons.all()]

	def to_dict(self, escape: bool = False) -> dict:
		return {
			'mode': self.mode,
			'buttons': self.get_buttons_as_dict(escape=escape),
		}

class TelegramBotCommandKeyboardButton(models.Model):
	telegram_bot_command_keyboard = models.ForeignKey(TelegramBotCommandKeyboard, on_delete=models.CASCADE, related_name='buttons', null=True)
	row = models.IntegerField(_('Ряд'), blank=True, null=True)
	text = models.TextField(_('Текст'), max_length=4096)
	url = models.TextField(_('URL-адрес'), max_length=2048, blank=True, null=True)

	telegram_bot_command = models.ForeignKey(TelegramBotCommand, on_delete=models.SET_NULL, blank=True, null=True)

	start_diagram_connector = models.TextField(blank=True, null=True)
	end_diagram_connector = models.TextField(blank=True, null=True)

	class Meta:
		db_table = 'telegram_bot_command_keyboard_button'
		ordering = ['id']

	def get_telegram_bot_command(self) -> Optional[TelegramBotCommand]:
		return self.telegram_bot_command

	async def aget_telegram_bot_command(self) -> Optional[TelegramBotCommand]:
		return await sync_to_async(self.get_telegram_bot_command)()

	def to_dict(self, escape: bool = False) -> dict:
		return {
			'id': self.id,
			'row': self.row,
			'text': filters.escape(self.text) if escape else self.text,
			'url': self.url,

			'telegram_bot_command_id': self.telegram_bot_command.id if self.telegram_bot_command else None,

			'start_diagram_connector': self.start_diagram_connector,
			'end_diagram_connector' : self.end_diagram_connector,
		}

class TelegramBotCommandApiRequest(models.Model):
	telegram_bot_command = models.OneToOneField(TelegramBotCommand, on_delete=models.CASCADE, related_name='api_request')
	url = models.TextField(_('URL-адрес'), max_length=2048)
	method = models.CharField(_('Метод'), max_length=6, choices=(
		('get', 'GET'),
		('post', 'POST'),
		('put', 'PUT'),
		('patch', 'PATCH'),
		('delete', 'DELETE'),
	), default='get')
	headers = models.JSONField(_('Заголовки'), blank=True, null=True)
	data = models.JSONField(_('Данные'), blank=True, null=True)

	class Meta:
		db_table = 'telegram_bot_command_api_request'

	def to_dict(self) -> dict:
		return {
			'url': self.url,
			'method': self.method,
			'headers': self.headers,
			'data': self.data,
		}

class TelegramBotUser(models.Model):
	telegram_bot = models.ForeignKey(TelegramBot, on_delete=models.CASCADE, related_name='users', verbose_name=_('Telegram бот'), null=True)
	user_id = models.BigIntegerField('Telegram ID')
	full_name = models.CharField(_('Полное имя'), max_length=129, null=True)
	is_allowed = models.BooleanField(_('Разрешён'), default=False)
	activated_date = models.DateTimeField(_('Дата активации'), auto_now_add=True)

	class Meta:
		db_table = 'telegram_bot_user'

		verbose_name = _('Пользователя')
		verbose_name_plural = _('Пользователи')

	def to_dict(self) -> dict:
		return {
			'id': self.id,
			'user_id': self.user_id,
			'full_name': self.full_name,
			'is_allowed': self.is_allowed,
			'activated_date': f'{filters.date(self.activated_date)} {filters.time(self.activated_date)}',
		}

	def __str__(self) -> str:
		return self.full_name

constructor_telegram_bots/telegram_bot/models.py
pyguru-dev/telegrambot
from django.utils.translation import gettext_lazy as _

from rest_framework import serializers

from user.models import User
from telegram_bot.models import TelegramBot

from .functions import check_telegram_bot_api_token

from typing import Optional


class CreateTelegramBotSerializer(serializers.Serializer):
	api_token = serializers.CharField(max_length=50, error_messages={
		'blank': _('Введите API-токен Telegram бота!'),
		'max_length': _('API-токен Telegram бота должен содержать не более 50 символов!'),
	})
	is_private = serializers.BooleanField()

	def validate_api_token(self, api_token: str) -> str:
		user: User = self.context['user']

		if user.telegram_bots.filter(api_token=api_token).exists():
			raise serializers.ValidationError(_('Вы уже используете этот API-токен Telegram бота на сайте!'))
		elif TelegramBot.objects.filter(api_token=api_token).exists():
			raise serializers.ValidationError(_('Этот API-токен Telegram бота уже использует другой пользователь сайта!'))

		if not check_telegram_bot_api_token(api_token):
			raise serializers.ValidationError(_('Ваш API-токен Telegram бота является недействительным!'))

		return api_token

class UpdateTelegramBotSerializer(CreateTelegramBotSerializer):
	def __init__(self, *args, **kwargs) -> None:
		super().__init__(*args, **kwargs)

		self.fields['api_token'].allow_null = True
		self.fields['is_private'].allow_null = True

	def validate_api_token(self, api_token: Optional[str]) -> Optional[str]:
		if api_token is not None:
			return super().validate_api_token(api_token)

		return api_token

class UpdateTelegramBotDiagramCurrentScaleSerializer(serializers.Serializer):
	diagram_current_scale = serializers.FloatField()

	def validate_diagram_current_scale(self, diagram_current_scale: float) -> float:
		if diagram_current_scale < 0.1 or diagram_current_scale > 2.0:
			return 1.0

		return diagram_current_scale

class TelegramBotCommandCommandSerializer(serializers.Serializer):
	text = serializers.CharField(max_length=32, error_messages={
		'blank': _('Введите команду!'),
		'max_length': _('Команда должна содержать не более 32 символов!'),
	})
	is_show_in_menu = serializers.BooleanField()
	description = serializers.CharField(max_length=255, error_messages={
		'blank': _('Введите описание для команды!'),
		'max_length': _('Описание команды должно содержать не более 255 символов!'),
	}, allow_null=True)

class TelegramBotCommandMessageTextSerializer(serializers.Serializer):
	mode = serializers.ChoiceField(choices=['default', 'markdown', 'html'])
	text = serializers.CharField(max_length=4096, error_messages={
		'blank': _('Введите текст сообщения!'),
		'max_length': _('Текст сообщения должен содержать не более 4096 символов!'),
	})

class TelegramBotCommandKeyboardButtonSerializer(serializers.Serializer):
	id = serializers.IntegerField(allow_null=True)
	row = serializers.IntegerField(allow_null=True)
	text = serializers.CharField(max_length=4096, error_messages={
		'blank': _('Введите название кнопки клавиатуры!'),
		'max_length': _('Название кнопки клавиатуры должно содержать не более 4096 символов!'),
	})
	url = serializers.URLField(max_length=2048, error_messages={
		'blank': _('Введите URL-адрес для кнопки клавиатуры!'),
		'invalid': _('Введите правильный URL-адрес для кнопки клавиатуры!'),
		'max_length': _('URL-адрес кнопки клавиатуры должен содержать не более 2048 символов!'),
	}, allow_null=True)

class TelegramBotCommandKeyboardSerializer(serializers.Serializer):
	mode =  serializers.ChoiceField(choices=['default', 'inline', 'payment'])
	buttons = TelegramBotCommandKeyboardButtonSerializer(many=True)

class TelegramBotCommandApiRequestSerializer(serializers.Serializer):
	url = serializers.URLField(max_length=2048, error_messages={
		'blank': _('Введите URL-адрес для API-запроса!'),
		'invalid': _('Введите правильный URL-адрес для API-запроса!'),
		'max_length': _('URL-адрес API-запроса должен содержать не более 2048 символов!'),
	})
	method = serializers.ChoiceField(choices=['get', 'post', 'put', 'patch', 'delete'])
	headers = serializers.JSONField(allow_null=True)
	data = serializers.JSONField(allow_null=True)

class CreateTelegramBotCommandSerializer(serializers.Serializer):
	name = serializers.CharField(max_length=255, error_messages={
		'blank': _('Введите название команде!'),
		'max_length': _('Название команды должно содержать не более 255 символов!'),
	})
	command = TelegramBotCommandCommandSerializer(allow_null=True)
	message_text = TelegramBotCommandMessageTextSerializer()
	keyboard = TelegramBotCommandKeyboardSerializer(allow_null=True)
	api_request = TelegramBotCommandApiRequestSerializer(allow_null=True)
	database_record = serializers.JSONField(allow_null=True)

class UpdateTelegramBotCommandSerializer(CreateTelegramBotCommandSerializer):
	pass

class UpdateTelegramBotCommandPositionSerializer(serializers.Serializer):
	x = serializers.IntegerField()
	y = serializers.IntegerField()

class UpdateTelegramBotCommandKeyboardButtonTelegramBotCommandSerializer(serializers.Serializer):
	telegram_bot_command_id = serializers.IntegerField()
	start_diagram_connector = serializers.CharField()
	end_diagram_connector = serializers.CharField()

	def validate_telegram_bot_command_id(self, telegram_bot_command_id: int) -> int:
		telegram_bot: TelegramBot = self.context['telegram_bot']

		if not telegram_bot.commands.filter(id=telegram_bot_command_id).exists():
			raise serializers.ValidationError(_('Команда Telegram бота не найдена!'))

		return telegram_bot_command_id

class CreateTelegramBotDatabeseRecordSerializer(serializers.Serializer):
	record = serializers.JSONField()

class UpdateTelegramBotDatabeseRecordSerializer(CreateTelegramBotDatabeseRecordSerializer):
	pass

constructor_telegram_bots/telegram_bot/serializers.py
pyguru-dev/telegrambot
from .telegram_bot import ConstructorTelegramBot
constructor_telegram_bots/telegram_bot/services/constructor_telegram_bot/__init__.py
pyguru-dev/telegrambot
from aiogram import types

from user.models import User, UserManager

from asgiref.sync import sync_to_async
from functools import wraps


def check_user(func):
	@wraps(func)
	async def wrapper(*args, **kwargs):
		message: types.Message = args[1]

		user: UserManager = await sync_to_async(User.objects.filter)(telegram_id=message.from_user.id)

		if not await user.aexists():
			await sync_to_async(User.objects.create)(telegram_id=message.from_user.id, first_name=message.from_user.first_name)

		return await func(*args, **kwargs)
	return wrapper

constructor_telegram_bots/telegram_bot/services/constructor_telegram_bot/decorators.py
pyguru-dev/telegrambot
from telegram_bot.services.custom_aiogram import CustomBot, CustomDispatcher
from aiogram import types

from .decorators import *

from user.models import User

from django.conf import settings
import asyncio


class ConstructorTelegramBot:
	def __init__(self) -> None:
		self.loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()

		self.bot = CustomBot(token=settings.CONSTRUCTOR_TELEGRAM_BOT_API_TOKEN, loop=self.loop)
		self.dispatcher = CustomDispatcher(bot_username=settings.CONSTRUCTOR_TELEGRAM_BOT_USERNAME, bot=self.bot)

	@check_user
	async def start_command(self, message: types.Message) -> None:
		await self.bot.send_message(chat_id=message.chat.id, text=f"""\
			Hello, @{message.from_user.username}!
			I am a Telegram bot for Constructor Telegram Bots site.
			Thank you for being with us ❤️
		""".replace('\t', ''))

		commands_list: list = message.text.split()

		if len(commands_list) > 1 and commands_list[1] == 'login':
			await self.login_command(message)

	@check_user
	async def login_command(self, message: types.Message) -> None:
		user: User = await User.objects.aget(telegram_id=message.from_user.id)

		keyboard = types.InlineKeyboardMarkup(row_width=1)
		keyboard.add(types.InlineKeyboardButton(text='Login', url=await user.alogin_url))

		await self.bot.send_message(chat_id=message.chat.id, text='Click on the button below to login on the site.', reply_markup=keyboard)

	async def setup(self) -> None:
		self.dispatcher.register_message_handler(self.start_command, commands=['start'])
		self.dispatcher.register_message_handler(self.login_command, commands=['login'])

	async def start(self) -> None:
		await self.dispatcher.skip_updates()
		await self.dispatcher.start_polling()

constructor_telegram_bots/telegram_bot/services/constructor_telegram_bot/telegram_bot.py
pyguru-dev/telegrambot
from telegram_bot.services.tests import BaseTestCase
from telegram_bot.services.custom_aiogram import CustomBot

from user.models import User

from .telegram_bot import ConstructorTelegramBot

from functools import wraps
import json


class ConstructorTelegramBotTests(BaseTestCase):
	def setUp(self) -> None:
		self.constructor_telegram_bot = ConstructorTelegramBot()

	def setup(func):
		wraps(func)
		async def wrapper(self, *args, **kwargs):
			await self.constructor_telegram_bot.setup()

			self.bot: CustomBot = self.constructor_telegram_bot.bot

			return await func(self, *args, **kwargs)
		return wrapper

	@setup
	async def test_start_command(self):
		result: dict = (await self.send_message(self.constructor_telegram_bot.start_command, '/start'))[0]

		assert result['method'] == 'sendMessage'
		assert result['text'] == f"""\
			Hello, @test!
			I am a Telegram bot for Constructor Telegram Bots site.
			Thank you for being with us ❤️
		""".replace('\t', '')

	@setup
	async def test_login_command(self):
		result: dict = (await self.send_message(self.constructor_telegram_bot.login_command, '/login'))[0]
		result_keyboard: dict = json.loads(result['reply_markup'])['inline_keyboard']

		user: User = await User.objects.afirst()
		login_url: str = await user.alogin_url

		assert result['method'] == 'sendMessage'
		assert result['text'] == 'Click on the button below to login on the site.'
		assert result_keyboard[0][0]['text'] == 'Login'
		assert result_keyboard[0][0]['url'] == login_url

constructor_telegram_bots/telegram_bot/services/constructor_telegram_bot/tests.py
pyguru-dev/telegrambot
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiohttp.helpers import sentinel
from aiogram.utils.exceptions import *

from django.conf import settings

import asyncio
import aiohttp

from typing import Optional, Union
import logging


log = logging.getLogger('aiogram')


class CustomBot(Bot):
	def __init__(self, *args, **kwargs) -> None:
		if settings.TEST:
			self.results = []
		else:
			super().__init__(*args, **kwargs)

	async def request(self, *args, **kwargs) ->  Union[list, dict, bool]:
		if settings.TEST:
			self.results.append({'method': args[0], **args[1]})
			return {}
		else:
			return await super().request(*args, **kwargs)

	async def get_results(self) -> Optional[list]:
		return self.results if settings.TEST else None

class CustomDispatcher(Dispatcher):
	def __init__(self, bot_username: str, bot: CustomBot):
		self.bot_username = bot_username

		super().__init__(bot)

	async def start_polling(self, timeout: int = 20, reset_webhook=None):
		log.info(f'@{self.bot_username} || Telegram bot is starting.')

		Dispatcher.set_current(self)
		Bot.set_current(self.bot)

		await self.reset_webhook(check=False if reset_webhook is None else True)

		self._polling = True
		current_request_timeout = self.bot.timeout
		offset = None

		if current_request_timeout is not sentinel and timeout is not None:
			request_timeout = aiohttp.ClientTimeout(total=current_request_timeout.total + timeout or 1)
		else:
			request_timeout = None

		while self._polling:
			try:
				with self.bot.request_timeout(request_timeout):
					updates: list[types.Update] = await self.bot.get_updates(offset=offset, timeout=timeout)

					for update in updates:
						try:
							log.info(f'@{self.bot_username} || {update.message.from_user.first_name}: {update.message.text}')
						except:
							pass
			except TerminatedByOtherGetUpdates:
				log.error(f'@{self.bot_username} || Telegram bot is already started!')
				break
			except (TelegramAPIError, ConflictError, NetworkError, asyncio.exceptions.TimeoutError):
				pass
			except asyncio.CancelledError:
				break
			except Exception as e:
				log.exception(f'@{self.bot_username} || Cause exception while getting updates!')
				break

			if updates:
				offset = updates[-1].update_id + 1

				asyncio.create_task(self._process_polling_updates(updates))

			await asyncio.sleep(0.1)

		self._close_waiter.set_result(None)

		log.info(f'@{self.bot_username} || Telegram bot is stopped.')

constructor_telegram_bots/telegram_bot/services/custom_aiogram.py
pyguru-dev/telegrambot
from pymongo import MongoClient
from pymongo.collection import Collection

from django.conf import settings

from typing import List, Union


def connect_to_database(func):
	def wrapper(telegram_bot, *args, **kwargs):
		client = MongoClient('127.0.0.1', 27017)
		collection = client.telegram_bots.get_collection('0' if settings.TEST else str(telegram_bot.id))
		result: Union[list, dict, None] = func(collection=collection, *args, **kwargs)
		client.close()

		return result
	return wrapper

@connect_to_database
def delete_collection(*, collection: Collection) -> None:
	collection.drop()

@connect_to_database
def insert_record(record: dict, *, collection: Collection) -> dict:
	if collection.count_documents({}) > 0:
		record_id = [record['_id'] for record in collection.find()][-1] + 1
	else:
		record_id = 1

	record.update({'_id': record_id})

	collection.insert_one(record)
	return collection.find_one({'_id': record['_id']})

@connect_to_database
def update_record(record_id: int, updated_record: dict, *, collection: Collection) -> dict:
	updated_record_ = {
		'$set': {},
		'$unset': {},
	}

	updated_record.update({'_id': record_id})
	old_record: dict = collection.find_one({'_id': record_id})

	for key, value in updated_record.items():
		if key in old_record and old_record[key] != value:
			updated_record_['$set'][key] = value
		else:
			updated_record_['$set'][key] = value

	for key in old_record:
		if key not in updated_record:
			updated_record_['$unset'][key] = 1

	collection.update_one({'_id': record_id}, updated_record_)
	return collection.find_one({'_id': record_id})

@connect_to_database
def delete_record(record_id: int, *, collection: Collection) -> None:
	collection.delete_one({'_id': record_id})

@connect_to_database
def get_records(*, collection: Collection) -> List[dict]:
	return [record for record in collection.find()]

constructor_telegram_bots/telegram_bot/services/database_telegram_bot/functions.py
pyguru-dev/telegrambot
from celery import shared_task

from .constructor_telegram_bot import ConstructorTelegramBot
from .user_telegram_bot import UserTelegramBot

from telegram_bot.models import TelegramBot

from threading import Thread
from typing import Union
from sys import platform


@shared_task
def stop_telegram_bot(telegram_bot_id: int) -> None:
	telegram_bot: TelegramBot = TelegramBot.objects.get(id=telegram_bot_id)
	telegram_bot.is_running = False
	telegram_bot.save()

async def start_telegram_bot__(telegram_bot: Union[ConstructorTelegramBot, UserTelegramBot]) -> None:
	await telegram_bot.setup()
	await telegram_bot.start()

def start_telegram_bot_(telegram_bot: Union[ConstructorTelegramBot, UserTelegramBot]) -> None:
	telegram_bot.loop.run_until_complete(start_telegram_bot__(telegram_bot))
	telegram_bot.loop.stop()

@shared_task
def start_telegram_bot(telegram_bot_id: int) -> None:
	telegram_bot: TelegramBot = TelegramBot.objects.get(id=telegram_bot_id)
	telegram_bot.is_running = True
	telegram_bot.is_stopped = False
	telegram_bot.save()

	user_telegram_bot = UserTelegramBot(telegram_bot=telegram_bot)
	Thread(target=start_telegram_bot_, args=(user_telegram_bot,), daemon=True).start()

@shared_task
def start_all_telegram_bots() -> None:
	constructor_telegram_bot = ConstructorTelegramBot()
	Thread(target=start_telegram_bot_, args=(constructor_telegram_bot,), daemon=True).start()

	for telegram_bot in TelegramBot.objects.all():
		if telegram_bot.is_running:
			if platform == 'win32':
				start_telegram_bot(telegram_bot_id=telegram_bot.id)
			else:
				start_telegram_bot.delay(telegram_bot_id=telegram_bot.id)

constructor_telegram_bots/telegram_bot/services/tasks.py
pyguru-dev/telegrambot
from django.test import TestCase

from aiogram import types


class BaseTestCase(TestCase):
	async def send_message(self, handler, text: str) -> list:
		message: types.Message = types.Update(**{
			'update_id': 1,
			'message': {
				'message_id': 1,
				'from': {
					'id': 1,
					'first_name': 'test',
					'username': 'test',
					'is_bot': False,
					'language_code': 'ru',
				},
				'chat': {
					'id': 1,
					'first_name': 'test',
					'username': 'test',
					'type': 'private',
				},
				'date': 1,
				'text': text,
			},
		}).message

		await handler(message)
		return await self.bot.get_results()

	async def send_callback_query(self, handler, data: str) -> list:
		callback_query: types.CallbackQuery = types.Update(**{
			'update_id': 1,
			'callback_query': {
				'id': '1',
				'from': {
					'id': 1,
					'first_name': 'Test',
					'username': 'Test',
					'is_bot': False,
					'language_code': 'ru',
				},
				'message': {
					'message_id': 1,
					'from': {
						'id': 1,
						'first_name': 'Test Telegram Bot',
						'is_bot': True,
						'username': 'test_bot',
					},
					'chat': {
						'id': 1,
						'first_name': 'Test',
						'username': 'Test',
						'type': 'private',
					},
					'date': 1,
					'text': '',
					'reply_markup': {
						'inline_keyboard': [],
					},
				},
				'chat_instance': '1',
				'data': data,
			},
		}).callback_query

		await handler(callback_query)
		return await self.bot.get_results()

constructor_telegram_bots/telegram_bot/services/tests.py
pyguru-dev/telegrambot
from .telegram_bot import UserTelegramBot
constructor_telegram_bots/telegram_bot/services/user_telegram_bot/__init__.py
pyguru-dev/telegrambot
from aiogram import types

from django.db import models

from constructor_telegram_bots import environment

from telegram_bot.models import *
from telegram_bot.services import database_telegram_bot

from .functions import search_telegram_bot_command, get_text_variables

from asgiref.sync import sync_to_async
import aiohttp

from functools import wraps
from typing import Optional, Union
import json


def check_telegram_bot_user(func):
	@wraps(func)
	async def wrapper(self, request: Union[types.Message, types.CallbackQuery], *args, **kwargs):
		telegram_bot: TelegramBot = self.telegram_bot
		telegram_bot_user_: models.Manager = await sync_to_async(telegram_bot.users.filter)(user_id=request.from_user.id)

		if not await telegram_bot_user_.aexists():
			telegram_bot_user: TelegramBotUser = await TelegramBotUser.objects.acreate(
				telegram_bot=telegram_bot,
				user_id=request.from_user.id,
				full_name=request.from_user.full_name
			)
		else:
			telegram_bot_user: TelegramBotUser = await telegram_bot_user_.afirst()

		if not telegram_bot.is_private or telegram_bot.is_private and telegram_bot_user.is_allowed:
			return await func(self, request, *args, **kwargs)
	return wrapper

def check_telegram_bot_command(func):
	@wraps(func)
	async def wrapper(self, request: Union[types.Message, types.CallbackQuery], *args, **kwargs):
		telegram_bot_command = None

		if isinstance(request, types.Message):
			text_variables: dict = await get_text_variables(telegram_bot=self.telegram_bot, request=request)

			async for telegram_bot_command_ in self.telegram_bot.commands.all():
				telegram_bot_command_command: Optional[TelegramBotCommandCommand] = await telegram_bot_command_.aget_command()

				if telegram_bot_command_command:
					telegram_bot_command_command_command: str = await environment.areplace_text_variables(
						telegram_bot=self.telegram_bot,
						text=telegram_bot_command_command.text,
						text_variables=text_variables
					)

					if telegram_bot_command_command_command == request.text:
						telegram_bot_command: TelegramBotCommand = telegram_bot_command_
				else:
					telegram_bot_command: Optional[TelegramBotCommand] = await search_telegram_bot_command(
						telegram_bot=self.telegram_bot,
						message_text=request.text
					)

				if telegram_bot_command:
					break
		elif isinstance(request, types.CallbackQuery) and request.data.isdigit():
			telegram_bot_command: Optional[TelegramBotCommand] = await search_telegram_bot_command(
				telegram_bot=self.telegram_bot,
				button_id=int(request.data)
			)

		if telegram_bot_command:
			return await func(self, request, telegram_bot_command=telegram_bot_command, *args, **kwargs)
	return wrapper

def check_telegram_bot_command_database_record(func):
	@wraps(func)
	async def wrapper(self, request: Union[types.Message, types.CallbackQuery], *args, **kwargs):
		telegram_bot_command: TelegramBotCommand = kwargs['telegram_bot_command']

		if telegram_bot_command.database_record:
			text_variables: dict = await get_text_variables(telegram_bot=self.telegram_bot, request=request)
			database_record: str = await environment.areplace_text_variables(
				telegram_bot=self.telegram_bot,
				text=telegram_bot_command.database_record,
				text_variables=text_variables
			)
			database_record: dict = json.loads(database_record)

			database_telegram_bot.insert_record(self.telegram_bot, database_record)
		return await func(self, request, *args, **kwargs)
	return wrapper

def check_message_text(func):
	@wraps(func)
	async def wrapper(self, request: Union[types.Message, types.CallbackQuery], *args, **kwargs):
		telegram_bot_command: TelegramBotCommand = kwargs['telegram_bot_command']
		telegram_bot_command_api_request: TelegramBotCommandApiRequest = await telegram_bot_command.aget_api_request()
		text_variables: dict = await get_text_variables(telegram_bot=self.telegram_bot, request=request)

		if telegram_bot_command_api_request:
			async with aiohttp.ClientSession() as session:
				method: str = telegram_bot_command_api_request.method
				url: str = await environment.areplace_text_variables(
					telegram_bot=self.telegram_bot,
					text=telegram_bot_command_api_request.url,
					text_variables=text_variables
				)

				if telegram_bot_command_api_request.headers:
					headers: str = await environment.areplace_text_variables(
						telegram_bot=self.telegram_bot,
						text=telegram_bot_command_api_request.headers,
						text_variables=text_variables
					)
					headers: dict = json.loads(headers)
				else:
					headers = None

				if telegram_bot_command_api_request.data:
					data: str = await environment.areplace_text_variables(
						telegram_bot=self.telegram_bot,
						text=telegram_bot_command_api_request.data,
						text_variables=text_variables
					)
					data: dict = json.loads(data)
				else:
					data = None

				try:
					async with session.request(method=method, url=url, headers=headers, json=data) as response:
						try:
							text_variables['api_response'] = await response.json()
						except aiohttp.client_exceptions.ContentTypeError:
							text_variables['api_response'] = 'API-request return not JSON!'
				except aiohttp.client_exceptions.InvalidURL:
					text_variables['api_response'] = 'URL is invalid!'

		telegram_bot_command_message_text: TelegramBotCommandMessageText = await telegram_bot_command.aget_message_text()
		message_text: str = await environment.areplace_text_variables(
			telegram_bot=self.telegram_bot,
			text=telegram_bot_command_message_text.text,
			text_variables=text_variables
		)

		if message_text:
			if len(message_text) > 4096:
				message_text = 'The message text must contain no more than 4096 characters!'
		else:
			message_text = 'The message text is empty!'

		return await func(self, request, message_text=message_text, *args, **kwargs)
	return wrapper

constructor_telegram_bots/telegram_bot/services/user_telegram_bot/decorators.py
pyguru-dev/telegrambot
from aiogram import types

from django.db import models

from telegram_bot.models import *
from telegram_bot.services import database_telegram_bot

from asgiref.sync import sync_to_async
from typing import Optional, Union


async def search_telegram_bot_command(
	telegram_bot: TelegramBot,
	message_text: Optional[str] = None,
	button_id: Optional[int] = None
) -> Optional[TelegramBotCommand]:
	if message_text:
		async for telegram_bot_command in telegram_bot.commands.all():
			telegram_bot_command_keyboard: TelegramBotCommandKeyboard = await telegram_bot_command.aget_keyboard()

			if telegram_bot_command_keyboard and telegram_bot_command_keyboard.mode == 'default':
				async for telegram_bot_command_keyboard_button in telegram_bot_command_keyboard.buttons.all():
					if telegram_bot_command_keyboard_button.text == message_text:
						return await telegram_bot_command_keyboard_button.aget_telegram_bot_command()
	elif button_id:
		telegram_bot_command_keyboard_button_: models.Manager = await sync_to_async(TelegramBotCommandKeyboardButton.objects.filter)(id=button_id)

		if await telegram_bot_command_keyboard_button_.aexists():
			telegram_bot_command_keyboard_button: TelegramBotCommandKeyboardButton = await telegram_bot_command_keyboard_button_.afirst()
			return await telegram_bot_command_keyboard_button.aget_telegram_bot_command()

async def get_text_variables(
	telegram_bot: TelegramBot,
	request: Union[types.Message, types.CallbackQuery]
) -> dict:
	database_records = {}

	for record in database_telegram_bot.get_records(telegram_bot):
		database_records[record['_id']] = record

	if isinstance(request, types.Message):
		user_message_id = request.message_id
		user_message_text = request.text
	else:
		user_message_id = request.message.message_id
		user_message_text = request.message.text

	text_variables = {
		'user_id': request.from_user.id,
		'user_username': request.from_user.username,
		'user_first_name': request.from_user.first_name,
		'user_last_name': request.from_user.last_name,
		'user_message_id': user_message_id,
		'user_message_text': user_message_text,
		'database_records': database_records,
	}

	return text_variables

async def get_keyboard(telegram_bot_command: TelegramBotCommand) -> Union[types.ReplyKeyboardMarkup, types.InlineKeyboardMarkup]:
	telegram_bot_command_keyboard: TelegramBotCommandKeyboard =  await sync_to_async(telegram_bot_command.get_keyboard)()

	if telegram_bot_command_keyboard:
		keyboard_buttons = {}
		keyboard_row = 1

		for num in range(await telegram_bot_command_keyboard.buttons.acount()):
			keyboard_buttons.update({num + 1: []})

		if telegram_bot_command_keyboard.mode == 'default':
			telegram_keyboard = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)

			async for button in telegram_bot_command_keyboard.buttons.all():
				keyboard_buttons[keyboard_row if not button.row else button.row].append(types.KeyboardButton(text=button.text))
				keyboard_row += 1
		else:
			telegram_keyboard = types.InlineKeyboardMarkup()

			async for button in telegram_bot_command_keyboard.buttons.all():
				keyboard_buttons[keyboard_row if not button.row else button.row].append(
					types.InlineKeyboardButton(
						text=button.text,
						url=button.url,
						callback_data=button.id
					)
				)
				keyboard_row += 1

		for keyboard_button in keyboard_buttons:
			telegram_keyboard.add(*keyboard_buttons[keyboard_button])
	else:
		telegram_keyboard = None

	return telegram_keyboard

constructor_telegram_bots/telegram_bot/services/user_telegram_bot/functions.py
pyguru-dev/telegrambot
from aiogram import types
from aiogram.utils.exceptions import *

from telegram_bot.services.custom_aiogram import CustomBot, CustomDispatcher

from telegram_bot.models import *

from .decorators import *
from .functions import get_keyboard

import asyncio
from aiohttp import ClientSession
from typing import Union


class UserTelegramBot:
	def __init__(self, telegram_bot: TelegramBot) -> None:
		self.telegram_bot = telegram_bot
		self.loop = asyncio.new_event_loop()

		self.bot = CustomBot(token=self.telegram_bot.api_token, loop=self.loop)
		self.dispatcher = CustomDispatcher(bot_username=self.telegram_bot.username, bot=self.bot)

	@check_telegram_bot_user
	@check_telegram_bot_command
	@check_telegram_bot_command_database_record
	@check_message_text
	async def message_and_callback_query_handler(
		self,
		request: Union[types.Message, types.CallbackQuery],
		telegram_bot_command: TelegramBotCommand,
		message_text: str
	) -> None:
		chat_id = (request if isinstance(request, types.Message) else request.message).chat.id
		telegram_bot_command_message_text: Optional[TelegramBotCommandMessageText] = await telegram_bot_command.aget_message_text()
		parse_mode = telegram_bot_command_message_text.mode if telegram_bot_command_message_text.mode != 'default' else None
		telegram_keyboard: Union[types.ReplyKeyboardMarkup, types.InlineKeyboardMarkup] = await get_keyboard(telegram_bot_command)

		async def send_answer(message_text: str):
			try:
				if isinstance(request, types.CallbackQuery):
					try:
						await self.dispatcher.bot.delete_message(
							chat_id=chat_id,
							message_id=request.message.message_id
						)
					except (MessageToDeleteNotFound, MessageCantBeDeleted):
						pass

				if telegram_bot_command.image:
					try:
						await self.dispatcher.bot.send_photo(
							chat_id=chat_id,
							photo=types.InputFile(telegram_bot_command.image.path),
							caption=message_text,
							parse_mode=parse_mode,
							reply_markup=telegram_keyboard
						)
						return
					except FileNotFoundError:
						telegram_bot_command.image = None
						await telegram_bot_command.asave()

				await self.dispatcher.bot.send_message(
					chat_id=chat_id,
					text=message_text,
					parse_mode=parse_mode,
					reply_markup=telegram_keyboard
				)
			except RetryAfter as exception:
				await asyncio.sleep(exception.timeout)
				await send_answer(message_text=message_text)

		await send_answer(message_text=message_text)

	async def setup(self) -> None:
		bot_commands = []

		async for telegram_bot_command in self.telegram_bot.commands.all():
			telegram_bot_command_command: TelegramBotCommandCommand = await telegram_bot_command.aget_command()

			if telegram_bot_command_command and telegram_bot_command_command.is_show_in_menu:
				bot_commands.append(types.BotCommand(
					command=telegram_bot_command_command.text.replace('/', ''),
					description=telegram_bot_command_command.description
				))

		await self.bot.set_my_commands(bot_commands)

		self.dispatcher.register_message_handler(self.message_and_callback_query_handler)
		self.dispatcher.register_callback_query_handler(self.message_and_callback_query_handler)

	async def start(self) -> None:
		task: asyncio.Task = self.loop.create_task(self.stop())

		try:
			await self.dispatcher.start_polling()

			self.telegram_bot.is_stopped = True
			await self.telegram_bot.asave()
		except (ValidationError, Unauthorized):
			await self.telegram_bot.adelete()

		task.cancel()

		session: ClientSession = await self.bot.get_session()
		await session.close()

	async def stop(self) -> None:
		while True:
			await self.telegram_bot.arefresh_from_db()

			if not self.telegram_bot.is_running:
				self.dispatcher.stop_polling()
				break

			await asyncio.sleep(5)

constructor_telegram_bots/telegram_bot/services/user_telegram_bot/telegram_bot.py
pyguru-dev/telegrambot
from telegram_bot.services.tests import BaseTestCase
from telegram_bot.services.custom_aiogram import CustomBot

from user.models import User
from telegram_bot.models import *

from .telegram_bot import UserTelegramBot

from asgiref.sync import sync_to_async
from functools import wraps
import json


class UserTelegramBotTests(BaseTestCase):
	def setUp(self) -> None:
		self.user: User = User.objects.create(telegram_id=123456789, first_name='exg1o')
		self.telegram_bot: TelegramBot = TelegramBot.objects.create(
			owner=self.user,
			api_token='123456789:qwertyuiop',
			is_private=False
		)

		self.user_telegram_bot = UserTelegramBot(telegram_bot=self.telegram_bot)
		self.handler = self.user_telegram_bot.message_and_callback_query_handler

	def setup(func):
		wraps(func)
		async def wrapper(self, *args, **kwargs):
			await self.user_telegram_bot.setup()

			self.bot: CustomBot = self.user_telegram_bot.bot

			return await func(self, *args, **kwargs)
		return wrapper

	@setup
	async def test_send_command(self) -> None:
		await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test',
			command={
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard=None,
			api_request=None,
			database_record=None
		)

		result: dict = (await self.send_message(self.handler, '/test'))[1]

		assert result['method'] == 'sendMessage'
		assert result['text'] == 'Test...'

	@setup
	async def test_send_command_with_default_keyboard(self) -> None:
		await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test',
			command={
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard={
				'mode': 'default',
				'buttons': [
					{
						'row': None,
						'text': '1',
						'url': None,
					},
					{
						'row': 2,
						'text': '2',
						'url': None,
					},
					{
						'row': 2,
						'text': '3',
						'url': None,
					},
				],
			},
			api_request=None,
			database_record=None
		)

		result: dict = (await self.send_message(self.handler, '/test'))[1]
		result_keyboard: dict = json.loads(result['reply_markup'])['keyboard']

		assert result['method'] == 'sendMessage'
		assert result['text'] == 'Test...'
		assert result_keyboard[0][0]['text'] == '1'
		assert result_keyboard[1][0]['text'] == '2'
		assert result_keyboard[1][1]['text'] == '3'

	@setup
	async def test_send_command_with_inline_keyboard(self) -> None:
		await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test',
			command={
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard={
				'mode': 'inline',
				'buttons': [
					{
						'row': None,
						'text': '1',
						'url': 'https://example.com/',
					},
				],
			},
			api_request=None,
			database_record=None
		)

		result: dict = (await self.send_message(self.handler, '/test'))[1]
		result_keyboard: dict = json.loads(result['reply_markup'])['inline_keyboard']

		assert result['method'] == 'sendMessage'
		assert result['text'] == 'Test...'
		assert result_keyboard[0][0]['text'] == '1'
		assert result_keyboard[0][0]['url'] == 'https://example.com/'

	@setup
	async def test_click_default_keyboard_button(self) -> None:
		telegram_bot_command_1: TelegramBotCommand = await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test',
			command={
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard={
				'mode': 'default',
				'buttons': [
					{
						'row': None,
						'text': '1',
						'url': None,
					},
				],
			},
			api_request=None,
			database_record=None
		)
		telegram_bot_command_2: TelegramBotCommand = await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test1',
			command=None,
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test1...',
			},
			keyboard=None,
			api_request=None,
			database_record=None
		)

		telegram_bot_command_keyboard: TelegramBotCommandKeyboard = await sync_to_async(telegram_bot_command_1.get_keyboard)()
		telegram_bot_command_keyboard_button: TelegramBotCommandKeyboardButton = await telegram_bot_command_keyboard.buttons.afirst()
		telegram_bot_command_keyboard_button.telegram_bot_command = telegram_bot_command_2
		await telegram_bot_command_keyboard_button.asave()

		result: dict = (await self.send_message(self.handler, '1'))[1]

		assert result['method'] == 'sendMessage'
		assert result['text'] == 'Test1...'

	@setup
	async def test_click_inline_keyboard_button(self) -> None:
		telegram_bot_command_1: TelegramBotCommand = await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test',
			command={
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard={
				'mode': 'inline',
				'buttons': [
					{
						'row': None,
						'text': '1',
						'url': None,
					},
				],
			},
			api_request=None,
			database_record=None
		)
		telegram_bot_command_2: TelegramBotCommand = await TelegramBotCommand.objects.acreate(
			telegram_bot=self.telegram_bot,
			name='Test1',
			command=None,
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test1...',
			},
			keyboard=None,
			api_request=None,
			database_record=None
		)

		telegram_bot_command_keyboard: TelegramBotCommandKeyboard = await sync_to_async(telegram_bot_command_1.get_keyboard)()
		telegram_bot_command_keyboard_button: TelegramBotCommandKeyboardButton = await telegram_bot_command_keyboard.buttons.afirst()
		telegram_bot_command_keyboard_button.telegram_bot_command = telegram_bot_command_2
		await telegram_bot_command_keyboard_button.asave()

		results: list = await self.send_callback_query(self.handler, '1')

		assert results[1]['method'] == 'deleteMessage'
		assert results[2]['method'] == 'sendMessage'
		assert results[2]['text'] == 'Test1...'

constructor_telegram_bots/telegram_bot/services/user_telegram_bot/tests.py
pyguru-dev/telegrambot
from django.test import TestCase, Client
from django.http import HttpResponse
from django import urls
from django.template import defaultfilters as filters

from user.models import User
from telegram_bot.models import *
from telegram_bot.services import database_telegram_bot


class BaseTestCase(TestCase):
	def setUp(self) -> None:
		self.client = Client(enforce_csrf_checks=True)
		self.user: User = User.objects.create(telegram_id=123456789, first_name='exg1o')
		self.telegram_bot: TelegramBot = TelegramBot.objects.create(
			owner=self.user,
			api_token='123456789:qwertyuiop',
			is_private=True
		)
		self.telegram_bot_command: TelegramBotCommand = TelegramBotCommand.objects.create(
			telegram_bot=self.telegram_bot,
			name='Test',
			command={
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard={
				'mode': 'default',
				'buttons': [
					{
						'row': 1,
						'text': '1',
						'url': None,
					},
					{
						'row': 1,
						'text': '2',
						'url': 'http://example.com/',
					},
				],
			},
			api_request={
				'url': 'http://example.com/',
				'method': 'get',
				'headers': None,
				'data': None,
			},
			database_record={'key': 'value'}
		)
		self.telegram_bot_user: TelegramBotUser = TelegramBotUser.objects.create(
			telegram_bot=self.telegram_bot,
			user_id=123456789,
			full_name='Test user'
		)

class TelegramBotModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot.owner, self.user)
		self.assertEqual(self.telegram_bot.username, '123456789:qwertyuiop_test_telegram_bot')
		self.assertEqual(self.telegram_bot.api_token, '123456789:qwertyuiop')
		self.assertTrue(self.telegram_bot.is_private)
		self.assertFalse(self.telegram_bot.is_running)
		self.assertTrue(self.telegram_bot.is_stopped)

	def test_get_commands_as_dict(self) -> None:
		telegram_bot_commands: list = self.telegram_bot.get_commands_as_dict()
		self.assertListEqual(telegram_bot_commands, [{
			'id': 1,
			'name': 'Test',
			'command': {
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			'image': None,
			'message_text': {
				'mode': 'default',
				'text': 'Test...',
			},
			'keyboard': {
				'mode': 'default',
				'buttons': [
					{
						'id': 1,
						'row': 1,
						'text': '1',
						'url': None,

						'telegram_bot_command_id': None,
						'start_diagram_connector': None,
						'end_diagram_connector' : None,
					},
					{
						'id': 2,
						'row': 1,
						'text': '2',
						'url': 'http://example.com/',

						'telegram_bot_command_id': None,
						'start_diagram_connector': None,
						'end_diagram_connector' : None,
					},
				],
			},
			'api_request': {
				'url': 'http://example.com/',
				'method': 'get',
				'headers': None,
				'data': None,
			},
			'database_record': {'key': 'value'},

			'x': 0,
			'y': 0,
		}])

	def test_get_users_as_dict(self) -> None:
		telegram_bot_users: list = self.telegram_bot.get_users_as_dict()
		self.assertListEqual(telegram_bot_users, [{
			'id': 1,
			'user_id': 123456789,
			'full_name': 'Test user',
			'is_allowed': False,
			'activated_date': f'{filters.date(self.telegram_bot_user.activated_date)} {filters.time(self.telegram_bot_user.activated_date)}',
		}])

	def test_to_dict(self) -> None:
		telegram_bot_as_dict: dict = self.telegram_bot.to_dict()
		self.assertDictEqual(telegram_bot_as_dict, {
			'id': 1,
			'username': '123456789:qwertyuiop_test_telegram_bot',
			'api_token': '123456789:qwertyuiop',
			'is_private': True,
			'is_running': False,
			'is_stopped': True,
			'commands_count': 1,
			'users_count': 1,
			'added_date': f'{filters.date(self.telegram_bot.added_date)} {filters.time(self.telegram_bot.added_date)}',
		})

class TelegramBotCommandModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_command.name, 'Test')
		self.assertEqual(self.telegram_bot_command.image, None)
		self.assertDictEqual(self.telegram_bot_command.database_record, {'key': 'value'})
		self.assertEqual(self.telegram_bot_command.x, 0)
		self.assertEqual(self.telegram_bot_command.y, 0)

	def test_get_command(self) -> None:
		telegram_bot_command_command: TelegramBotCommandCommand = self.telegram_bot_command.get_command()
		self.assertEqual(telegram_bot_command_command, self.telegram_bot_command.command)

	def test_get_keyboard(self) -> None:
		telegram_bot_command_keyboard: TelegramBotCommandKeyboard = self.telegram_bot_command.get_keyboard()
		self.assertEqual(telegram_bot_command_keyboard, self.telegram_bot_command.keyboard)

	def test_get_keyboard_as_dict(self) -> None:
		telegram_bot_command_keyboard: dict = self.telegram_bot_command.get_keyboard_as_dict()
		self.assertDictEqual(telegram_bot_command_keyboard, {
			'mode': 'default',
			'buttons': [
				{
					'id': 1,
					'row': 1,
					'text': '1',
					'url': None,

					'telegram_bot_command_id': None,
					'start_diagram_connector': None,
					'end_diagram_connector' : None,
				},
				{
					'id': 2,
					'row': 1,
					'text': '2',
					'url': 'http://example.com/',

					'telegram_bot_command_id': None,
					'start_diagram_connector': None,
					'end_diagram_connector' : None,
				},
			],
		})

	def test_get_api_request(self) -> None:
		telegram_bot_command_api_request: TelegramBotCommandApiRequest = self.telegram_bot_command.get_api_request()
		self.assertEqual(telegram_bot_command_api_request, self.telegram_bot_command.api_request)

	def test_to_dict(self) -> None:
		telegram_bot_command: dict = self.telegram_bot_command.to_dict()
		self.assertDictEqual(telegram_bot_command, {
			'id': 1,
			'name': 'Test',
			'command': {
				'text': '/test',
				'is_show_in_menu': False,
				'description': None,
			},
			'image': None,
			'message_text': {
				'mode': 'default',
				'text': 'Test...',
			},
			'keyboard': {
				'mode': 'default',
				'buttons': [
					{
						'id': 1,
						'row': 1,
						'text': '1',
						'url': None,

						'telegram_bot_command_id': None,
						'start_diagram_connector': None,
						'end_diagram_connector' : None,
					},
					{
						'id': 2,
						'row': 1,
						'text': '2',
						'url': 'http://example.com/',

						'telegram_bot_command_id': None,
						'start_diagram_connector': None,
						'end_diagram_connector' : None,
					},
				],
			},
			'api_request': {
				'url': 'http://example.com/',
				'method': 'get',
				'headers': None,
				'data': None,
			},
			'database_record': {'key': 'value'},

			'x': 0,
			'y': 0,
		})

class TelegramBotCommandCommandModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_command.command.text, '/test')
		self.assertFalse(self.telegram_bot_command.command.is_show_in_menu)
		self.assertIsNone(self.telegram_bot_command.command.description)

	def test_to_dict(self) -> None:
		telegram_bot_command_command: dict = self.telegram_bot_command.command.to_dict()
		self.assertDictEqual(telegram_bot_command_command, {
			'text': '/test',
			'is_show_in_menu': False,
			'description': None,
		})

class TelegramBotCommandMessageTextModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_command.message_text.mode, 'default')
		self.assertEqual(self.telegram_bot_command.message_text.text, 'Test...')

	def test_to_dict(self) -> None:
		telegram_bot_command_message_text: dict = self.telegram_bot_command.message_text.to_dict()
		self.assertDictEqual(telegram_bot_command_message_text, {
			'mode': 'default',
			'text': 'Test...',
		})

class TelegramBotCommandKeyboardModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_command.keyboard.mode, 'default')

	def test_get_buttons_as_dict(self) -> None:
		telegram_bot_command_keyboard_buttons: list = self.telegram_bot_command.keyboard.get_buttons_as_dict()
		self.assertListEqual(telegram_bot_command_keyboard_buttons, [
			{
				'id': 1,
				'row': 1,
				'text': '1',
				'url': None,

				'telegram_bot_command_id': None,
				'start_diagram_connector': None,
				'end_diagram_connector' : None,
			},
			{
				'id': 2,
				'row': 1,
				'text': '2',
				'url': 'http://example.com/',

				'telegram_bot_command_id': None,
				'start_diagram_connector': None,
				'end_diagram_connector' : None,
			},
		])

	def test_to_dict(self) -> None:
		telegram_bot_command_keyboard: dict = self.telegram_bot_command.keyboard.to_dict()
		self.assertDictEqual(telegram_bot_command_keyboard, {
			'mode': 'default',
			'buttons': [
				{
					'id': 1,
					'row': 1,
					'text': '1',
					'url': None,

					'telegram_bot_command_id': None,
					'start_diagram_connector': None,
					'end_diagram_connector' : None,
				},
				{
					'id': 2,
					'row': 1,
					'text': '2',
					'url': 'http://example.com/',

					'telegram_bot_command_id': None,
					'start_diagram_connector': None,
					'end_diagram_connector' : None,
				},
			],
		})

class TelegramBotCommandKeyboardButtonModelTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.telegram_bot_command_keyboard_button: TelegramBotCommandKeyboardButton = self.telegram_bot_command.keyboard.buttons.all()[0]

	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_command_keyboard_button.row, 1)
		self.assertEqual(self.telegram_bot_command_keyboard_button.text, '1')
		self.assertIsNone(self.telegram_bot_command_keyboard_button.url)
		self.assertIsNone(self.telegram_bot_command_keyboard_button.telegram_bot_command)
		self.assertIsNone(self.telegram_bot_command_keyboard_button.start_diagram_connector)
		self.assertIsNone(self.telegram_bot_command_keyboard_button.end_diagram_connector)

	def test_to_dict(self) -> None:
		telegram_bot_command_keyboard_button: dict = self.telegram_bot_command_keyboard_button.to_dict()
		self.assertDictEqual(telegram_bot_command_keyboard_button, {
			'id': 1,
			'row': 1,
			'text': '1',
			'url': None,

			'telegram_bot_command_id': None,
			'start_diagram_connector': None,
			'end_diagram_connector' : None,
		})

class TelegramBotCommandApiRequestModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_command.api_request.url, 'http://example.com/')
		self.assertEqual(self.telegram_bot_command.api_request.method, 'get')
		self.assertIsNone(self.telegram_bot_command.api_request.headers)
		self.assertIsNone(self.telegram_bot_command.api_request.data)

	def test_to_dict(self) -> None:
		telegram_bot_command_api_request: dict = self.telegram_bot_command.api_request.to_dict()
		self.assertDictEqual(telegram_bot_command_api_request, {
			'url': 'http://example.com/',
			'method': 'get',
			'headers': None,
			'data': None,
		})

class TelegramBotUserModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.telegram_bot_user.user_id, 123456789)
		self.assertEqual(self.telegram_bot_user.full_name, 'Test user')
		self.assertEqual(self.telegram_bot_user.is_allowed, False)

	def test_to_dict(self) -> None:
		telegram_bot_user: dict = self.telegram_bot_user.to_dict()
		self.assertDictEqual(telegram_bot_user, {
			'id': 1,
			'user_id': 123456789,
			'full_name': 'Test user',
			'is_allowed': False,
			'activated_date': f'{filters.date(self.telegram_bot_user.activated_date)} {filters.time(self.telegram_bot_user.activated_date)}',
		})

class TelegramBotsViewTests(BaseTestCase):
	url: str = urls.reverse('telegram_bots')

	def test_post_method(self) -> None:
		response: HttpResponse = self.client.post(self.url)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(
			self.url,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': '',
				'is_private': False,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': '123456789:qwertyuiop',
				'is_private': False,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.post(
			self.url,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': '123456789:dwawdadwa',
				'is_private': False,
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(self.url, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, self.user.get_telegram_bots_as_dict())

class TelegramBotViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('telegram_bot', kwargs={'telegram_bot_id': 0})

	def test_patch_method(self) -> None:
		response: HttpResponse = self.client.patch(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.patch(
			self.url_2,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': None,
				'is_private': None,
			}
		)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': None,
				'is_private': None,
			}
		)
		self.assertEqual(response.status_code, 500)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': '',
				'is_private': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': '123456789:qwertyuiop',
				'is_private': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': '123456789:dwawdadwa',
				'is_private': None,
			}
		)
		self.assertEqual(response.status_code, 200)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': None,
				'is_private': False,
			}
		)
		self.assertEqual(response.status_code, 200)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'api_token': None,
				'is_private': True,
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_delete_method(self) -> None:
		response: HttpResponse = self.client.delete(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.delete(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, self.telegram_bot.to_dict())

class StartOrStopTelegramBotViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('start_or_stop_telegram_bot', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('start_or_stop_telegram_bot', kwargs={'telegram_bot_id': 0})

	def test_start_or_stop_telegram_bot_view(self) -> None:
		response: HttpResponse = self.client.post(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

class UpdateTelegramBotDiagramCurrentScaleViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('update_telegram_bot_diagram_current_scale', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('update_telegram_bot_diagram_current_scale', kwargs={'telegram_bot_id': 0})

	def test_patch_method(self) -> None:
		response: HttpResponse = self.client.patch(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.patch(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={'diagram_current_scale': 0.8}
		)
		self.assertEqual(response.status_code, 200)

class TelegramBotCommandsViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_commands', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('telegram_bot_commands', kwargs={'telegram_bot_id': 0})

	def test_post_method(self) -> None:
		response: HttpResponse = self.client.post(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': '',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': '',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': {
					'text': '',
					'is_show_in_menu': False,
					'description': None
				},
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': {
					'text': '/start',
					'is_show_in_menu': True,
					'description': ''
				},
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': '',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': {
					'mode': 'default',
					'buttons': [
						{
							'id': None,
							'row': None,
							'text': '',
							'url': None,
						},
					],
				},
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': {
					'mode': 'default',
					'buttons': [
						{
							'id': None,
							'row': None,
							'text': '1',
							'url': '',
						},
					],
				},
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': {
					'mode': 'default',
					'buttons': [
						{
							'id': None,
							'row': None,
							'text': '1',
							'url': '-',
						},
					],
				},
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': {
					'url': '',
					'method': 'get',
					'headers': None,
					'data': None,
				},
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': {
					'url': 'test',
					'method': 'get',
					'headers': None,
					'data': None,
				},
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, self.telegram_bot.get_commands_as_dict(escape=True))

class TelegramBotCommandViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_command', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': self.telegram_bot_command.id,
		})
		self.url_2: str = urls.reverse('telegram_bot_command', kwargs={
			'telegram_bot_id': 0,
			'telegram_bot_command_id': self.telegram_bot_command.id,
		})
		self.url_3: str = urls.reverse('telegram_bot_command', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': 0,
		})

	def test_patch_method(self) -> None:
		response: HttpResponse = self.client.patch(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.patch(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': '',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': '',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': {
					'text': '',
					'is_show_in_menu': False,
					'description': None,
				},
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': {
					'text': '/start',
					'is_show_in_menu': True,
					'description': '',
				},
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': {
					'mode': 'default',
					'buttons': [
						{
							'id': '1',
							'row': None,
							'text': '',
							'url': None,
						},
					],
				},
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': {
					'mode': 'default',
					'buttons': [
						{
							'id': '1',
							'row': None,
							'text': '1',
							'url': '',
						},
					],
				},
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': {
					'mode': 'default',
					'buttons': [
						{
							'id': '1',
							'row': None,
							'text': '1',
							'url': '-',
						},
					],
				},
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': {
					'url': '',
					'method': 'get',
					'headers': None,
					'data': None,
				},
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'blank')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': {
					'url': 'test',
					'method': 'get',
					'headers': None,
					'data': None,
				},
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 400)
		self.assertEqual(response.json()['code'], 'invalid')

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'name': 'Test',
				'command': None,
				'message_text': {
					'mode': 'default',
					'text': 'Test...',
				},
				'keyboard': None,
				'api_request': None,
				'database_record': None,
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_delete_method(self) -> None:
		response: HttpResponse = self.client.delete(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.delete(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, self.telegram_bot_command.to_dict())

class UpdateTelegramBotCommandPositionViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('update_telegram_bot_command_position', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': self.telegram_bot_command.id,
		})
		self.url_2: str = urls.reverse('update_telegram_bot_command_position', kwargs={
			'telegram_bot_id': 0,
			'telegram_bot_command_id': self.telegram_bot_command.id,
		})
		self.url_3: str = urls.reverse('update_telegram_bot_command_position', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': 0,
		})

	def test_patch_method(self) -> None:
		response: HttpResponse = self.client.patch(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.patch(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'},)
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.patch(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'x': 123,
				'y': 321,
			}
		)
		self.assertEqual(response.status_code, 200)

class TelegramBotCommandKeyboardButtonTelegramBotCommandViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		telegram_bot_command_keyboard_button_id: int = self.telegram_bot_command.keyboard.buttons.all()[0].id

		self.url_1: str = urls.reverse('telegram_bot_command_keyboard_button_telegram_bot_command', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': self.telegram_bot_command.id,
			'telegram_bot_command_keyboard_button_id': telegram_bot_command_keyboard_button_id,
		})
		self.url_2: str = urls.reverse('telegram_bot_command_keyboard_button_telegram_bot_command', kwargs={
			'telegram_bot_id': 0,
			'telegram_bot_command_id': self.telegram_bot_command.id,
			'telegram_bot_command_keyboard_button_id': telegram_bot_command_keyboard_button_id,
		})
		self.url_3: str = urls.reverse('telegram_bot_command_keyboard_button_telegram_bot_command', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': 0,
			'telegram_bot_command_keyboard_button_id': telegram_bot_command_keyboard_button_id,
		})
		self.url_4: str = urls.reverse('telegram_bot_command_keyboard_button_telegram_bot_command', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_command_id': self.telegram_bot_command.id,
			'telegram_bot_command_keyboard_button_id': 0,
		})

	def test_post_method(self) -> None:
		response: HttpResponse = self.client.post(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(self.url_4, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(
			self.url_3,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'telegram_bot_command_id': 0,
				'start_diagram_connector': 'test',
				'end_diagram_connector': 'test',
			}
		)
		self.assertEqual(response.status_code, 404)

		telegram_bot_command: TelegramBotCommand = TelegramBotCommand.objects.create(
			telegram_bot=self.telegram_bot,
			name='Test',
			command=None,
			image=None,
			message_text={
				'mode': 'default',
				'text': 'Test...',
			},
			keyboard = None,
			api_request = None,
			database_record = None
		)

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={
				'telegram_bot_command_id': telegram_bot_command.id,
				'start_diagram_connector': 'test',
				'end_diagram_connector': 'test',
			}
		)
		self.assertEqual(response.status_code, 200)

	def test_delete_method(self) -> None:
		response: HttpResponse = self.client.delete(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.delete(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_4, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

class TelegramBotUsersViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_users', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('telegram_bot_users', kwargs={'telegram_bot_id': 0})

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.get(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, self.telegram_bot.get_users_as_dict())

class TelegramBotUserViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_user', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_user_id': self.telegram_bot_user.id,
		})
		self.url_2: str = urls.reverse('telegram_bot_user', kwargs={
			'telegram_bot_id': 0,
			'telegram_bot_user_id': self.telegram_bot_user.id,
		})
		self.url_3: str = urls.reverse('telegram_bot_user', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_user_id': 0,
		})

	def test_delete_method(self) -> None:
		response: HttpResponse = self.client.delete(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.delete(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

class TelegramBotAllowedUserViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_allowed_user', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_user_id': self.telegram_bot_user.id,
		})
		self.url_2: str = urls.reverse('telegram_bot_allowed_user', kwargs={
			'telegram_bot_id': 0,
			'telegram_bot_user_id': self.telegram_bot_user.id,
		})
		self.url_3: str = urls.reverse('telegram_bot_allowed_user', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'telegram_bot_user_id': 0,
		})

	def test_post_method(self) -> None:
		response: HttpResponse = self.client.post(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

	def test_delete_method(self) -> None:
		response: HttpResponse = self.client.delete(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.delete(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_3, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.delete(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)

class TelegramBotDatabeseRecordsViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_databese_records', kwargs={'telegram_bot_id': self.telegram_bot.id})
		self.url_2: str = urls.reverse('telegram_bot_databese_records', kwargs={'telegram_bot_id': 0})

	def tearDown(self) -> None:
		database_telegram_bot.delete_collection(self.telegram_bot)

	def test_post_method(self) -> None:
		response: HttpResponse = self.client.post(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.post(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		response: HttpResponse = self.client.post(
			self.url_1,
			headers={'Authorization': f'Token {self.user.auth_token.key}'},
			content_type='application/json',
			data={'record': {'key': 'value'}})
		self.assertEqual(response.status_code, 200)

	def test_get_method(self) -> None:
		response: HttpResponse = self.client.get(self.url_1)
		self.assertEqual(response.status_code, 401)

		response: HttpResponse = self.client.get(self.url_2, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 404)

		database_telegram_bot.insert_record(self.telegram_bot, {'key': 'value'})

		response: HttpResponse = self.client.get(self.url_1, headers={'Authorization': f'Token {self.user.auth_token.key}'})
		self.assertEqual(response.status_code, 200)
		self.assertJSONEqual(response.content, [{'_id': 1, 'key': 'value'}])

class TelegramBotDatabeseRecordViewTests(BaseTestCase):
	def setUp(self) -> None:
		super().setUp()

		self.url_1: str = urls.reverse('telegram_bot_databese_record', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'record_id': 1,
		})
		self.url_2: str = urls.reverse('telegram_bot_databese_record', kwargs={
			'telegram_bot_id': 0,
			'record_id': 0,
		})
		self.url_3: str = urls.reverse('telegram_bot_databese_record', kwargs={
			'telegram_bot_id': self.telegram_bot.id,
			'record_id': 0,
		})

	def tearDown(self) -> None:
		database_telegram_bot.delete_collection(self.telegram_bot)

	def test_patch_method(self) -> None:
		# Надо дописать
		pass

	def test_delete_method(self) -> None:
		# Надо дописать
		pass

constructor_telegram_bots/telegram_bot/tests.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.TelegramBotsView.as_view(), name='telegram_bots'), # POST, GET
    path('<int:telegram_bot_id>/', views.TelegramBotView.as_view(), name='telegram_bot'), # PATCH, DELETE, GET
	path('<int:telegram_bot_id>/start_or_stop/', views.start_or_stop_telegram_bot, name='start_or_stop_telegram_bot'), # POST
    path('<int:telegram_bot_id>/update-diagram-current-scale/', views.update_telegram_bot_diagram_current_scale, name='update_telegram_bot_diagram_current_scale'), # PATCH

	path('<int:telegram_bot_id>/commands/', views.TelegramBotCommandsView.as_view(), name='telegram_bot_commands'), # POST, GET
	path('<int:telegram_bot_id>/commands/<int:telegram_bot_command_id>/', views.TelegramBotCommandView.as_view(), name='telegram_bot_command'), # PATCH, DELETE, GET
	path('<int:telegram_bot_id>/commands/<int:telegram_bot_command_id>/update-position/', views.update_telegram_bot_command_position, name='update_telegram_bot_command_position'), # PATCH
	path('<int:telegram_bot_id>/commands/<int:telegram_bot_command_id>/keyboard-buttons/<int:telegram_bot_command_keyboard_button_id>/telegram-bot-command/', views.TelegramBotCommandKeyboardButtonTelegramBotCommandView.as_view(), name='telegram_bot_command_keyboard_button_telegram_bot_command'), # POST, DELETE

	path('<int:telegram_bot_id>/users/', views.TelegramBotUsersView.as_view(), name='telegram_bot_users'), # GET
	path('<int:telegram_bot_id>/users/<int:telegram_bot_user_id>/', views.TelegramBotUserView.as_view(), name='telegram_bot_user'), # DELETE
	path('<int:telegram_bot_id>/users/<int:telegram_bot_user_id>/allowed-user/', views.TelegramBotAllowedUserView.as_view(), name='telegram_bot_allowed_user'), # POST, DELETE

	path('<int:telegram_bot_id>/database/records/', views.TelegramBotDatabeseRecordsView.as_view(), name='telegram_bot_databese_records'), # POST, GET
    path('<int:telegram_bot_id>/database/records/<int:record_id>/', views.TelegramBotDatabeseRecordView.as_view(), name='telegram_bot_databese_record'), # PATCH, DELETE
]

constructor_telegram_bots/telegram_bot/urls.py
pyguru-dev/telegrambot
from django.utils.translation import gettext as _
from django.conf import settings

from rest_framework.views import APIView
from rest_framework.decorators import api_view, authentication_classes, permission_classes
from rest_framework.authentication import TokenAuthentication
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.status import *

from .models import *
from .services import tasks, database_telegram_bot
from .decorators import *
from .functions import *
from .serializers import *

from typing import Optional
import json
import sys


class TelegramBotsView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	def post(self, request: Request) -> Response:
		serializer = CreateTelegramBotSerializer(data=request.data, context={'user': request.user})
		serializer.is_valid(raise_exception=True)

		validated_data: dict = serializer.validated_data

		telegram_bot: TelegramBot = TelegramBot.objects.create(
			owner=request.user,
			api_token=validated_data['api_token'],
			is_private=validated_data['is_private']
		)

		return Response({
			'message': _('Вы успешно добавили Telegram бота.'),
			'level': 'success',

			'telegram_bot': telegram_bot.to_dict(),
		})

	def get(self, request: Request) -> Response:
		return Response(request.user.get_telegram_bots_as_dict())

class TelegramBotView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	def patch(self, request: Request, telegram_bot: TelegramBot) -> Response:
		serializer = UpdateTelegramBotSerializer(data=request.data, context={'user': request.user})
		serializer.is_valid(raise_exception=True)

		validated_data: dict = serializer.validated_data
		api_token: Optional[str] = validated_data['api_token']
		is_private: Optional[bool] = validated_data['is_private']

		if api_token is not None:
			username: str = check_telegram_bot_api_token(api_token)

			telegram_bot.username = username
			telegram_bot.api_token = api_token
			telegram_bot.is_running = False
			telegram_bot.save()

			return Response({
				'message': _('Вы успешно изменили API-токен Telegram бота.'),
				'level': 'success',
			})
		elif is_private is not None:
			telegram_bot.is_private = is_private
			telegram_bot.save()

			if is_private:
				return Response({
					'message': _('Вы успешно сделали Telegram бота приватным.'),
					'level': 'success',
				})
			else:
				return Response({
					'message': _('Вы успешно сделали Telegram бота не приватным.'),
					'level': 'success',
				})
		else:
			return Response({
				'message': _('На стороне сайта произошла непредвиденная ошибка, попробуйте ещё раз позже!'),
				'level': 'danger',
			}, status=HTTP_500_INTERNAL_SERVER_ERROR)

	@check_telegram_bot_id
	def delete(self, request: Request, telegram_bot: TelegramBot) -> Response:
		telegram_bot.delete()

		return Response({
			'message': _('Вы успешно удалили Telegram бота.'),
			'level': 'success',
		})

	@check_telegram_bot_id
	def get(self, request: Request, telegram_bot: TelegramBot) -> Response:
		return Response(telegram_bot.to_dict())

@api_view(['POST'])
@authentication_classes([TokenAuthentication])
@permission_classes([IsAuthenticated])
@check_telegram_bot_id
def start_or_stop_telegram_bot(request: Request, telegram_bot: TelegramBot) -> Response:
	if not settings.TEST:
		if not telegram_bot.is_running and telegram_bot.is_stopped:
			if sys.platform == 'win32':
				tasks.start_telegram_bot(telegram_bot_id=telegram_bot.id)
			else:
				tasks.start_telegram_bot.delay(telegram_bot_id=telegram_bot.id)
		elif telegram_bot.is_running and not telegram_bot.is_stopped:
			if sys.platform == 'win32':
				tasks.stop_telegram_bot(telegram_bot_id=telegram_bot.id)
			else:
				tasks.stop_telegram_bot.delay(telegram_bot_id=telegram_bot.id)

	return Response({
		'message': None,
		'level': 'success',
	})

@api_view(['PATCH'])
@authentication_classes([TokenAuthentication])
@permission_classes([IsAuthenticated])
@check_telegram_bot_id
def update_telegram_bot_diagram_current_scale(request: Request, telegram_bot: TelegramBot) -> Response:
	serializer = UpdateTelegramBotDiagramCurrentScaleSerializer(data=request.data)
	serializer.is_valid(raise_exception=True)

	telegram_bot.diagram_current_scale = serializer.validated_data['diagram_current_scale']
	telegram_bot.save()

	return Response({
		'message': None,
		'level': 'success',
	})

class TelegramBotCommandsView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	def post(self, request: Request, telegram_bot: TelegramBot) -> Response:
		request_data: dict = json.loads(request.POST['data']) if 'data' in request.POST else request.data

		serializer = CreateTelegramBotCommandSerializer(data=request_data)
		serializer.is_valid(raise_exception=True)

		TelegramBotCommand.objects.create(
			telegram_bot=telegram_bot,
			image=get_image_from_request(request),
			**serializer.validated_data
		)

		return Response({
			'message': _('Вы успешно добавили команду Telegram боту.'),
			'level': 'success',
		})

	@check_telegram_bot_id
	def get(self, request: Request, telegram_bot: TelegramBot) -> Response:
		return Response(telegram_bot.get_commands_as_dict(escape=True))

class TelegramBotCommandView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	@check_telegram_bot_command_id
	def patch(self, request: Request, telegram_bot: TelegramBot, telegram_bot_command: TelegramBotCommand) -> Response:
		request_data: dict = json.loads(request.POST['data']) if 'data' in request.POST else request.data

		serializer = UpdateTelegramBotCommandSerializer(data=request_data)
		serializer.is_valid(raise_exception=True)

		telegram_bot_command.update(
			image=get_image_from_request(request),
			**serializer.validated_data
		)

		return Response({
			'message': _('Вы успешно изменили команду Telegram бота.'),
			'level': 'success',
		})

	@check_telegram_bot_id
	@check_telegram_bot_command_id
	def delete(self, request: Request, telegram_bot: TelegramBot, telegram_bot_command: TelegramBotCommand) -> Response:
		telegram_bot_command.delete()

		return Response({
			'message': _('Вы успешно удалили команду Telegram бота.'),
			'level': 'success',
		})

	@check_telegram_bot_id
	@check_telegram_bot_command_id
	def get(self, request: Request, telegram_bot: TelegramBot, telegram_bot_command: TelegramBotCommand) -> Response:
		return Response(telegram_bot_command.to_dict())

@api_view(['PATCH'])
@authentication_classes([TokenAuthentication])
@permission_classes([IsAuthenticated])
@check_telegram_bot_id
@check_telegram_bot_command_id
def update_telegram_bot_command_position(request: Request, telegram_bot: TelegramBot, telegram_bot_command: TelegramBotCommand) -> Response:
	serializer = UpdateTelegramBotCommandPositionSerializer(data=request.data)
	serializer.is_valid(raise_exception=True)

	validated_data: dict = serializer.validated_data

	telegram_bot_command.x = validated_data['x']
	telegram_bot_command.y = validated_data['y']
	telegram_bot_command.save()

	return Response({
		'message': None,
		'level': 'success',
	})

class TelegramBotCommandKeyboardButtonTelegramBotCommandView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	@check_telegram_bot_command_id
	@check_telegram_bot_command_keyboard_button_id
	def post(
		self,
		request: Request,
		telegram_bot: TelegramBot,
		telegram_bot_command: TelegramBotCommand,
		telegram_bot_command_keyboard_button: TelegramBotCommandKeyboardButton,
	) -> Response:
		serializer = UpdateTelegramBotCommandKeyboardButtonTelegramBotCommandSerializer(data=request.data, context={'telegram_bot': telegram_bot})
		serializer.is_valid(raise_exception=True)

		validated_data: dict = serializer.validated_data

		telegram_bot_command_keyboard_button.telegram_bot_command = telegram_bot.commands.get(id=validated_data['telegram_bot_command_id'])
		telegram_bot_command_keyboard_button.start_diagram_connector = validated_data['start_diagram_connector']
		telegram_bot_command_keyboard_button.end_diagram_connector = validated_data['end_diagram_connector']
		telegram_bot_command_keyboard_button.save()

		return Response({
			'message': None,
			'level': 'success',
		})

	@check_telegram_bot_id
	@check_telegram_bot_command_id
	@check_telegram_bot_command_keyboard_button_id
	def delete(
		self,
		request: Request,
		telegram_bot: TelegramBot,
		telegram_bot_command: TelegramBotCommand,
		telegram_bot_command_keyboard_button: TelegramBotCommandKeyboardButton
	) -> Response:
		telegram_bot_command_keyboard_button.telegram_bot_command = None
		telegram_bot_command_keyboard_button.start_diagram_connector = None
		telegram_bot_command_keyboard_button.end_diagram_connector = None
		telegram_bot_command_keyboard_button.save()

		return Response({
			'message': None,
			'level': 'success',
		})

class TelegramBotUsersView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	def get(self, request: Request, telegram_bot: TelegramBot) -> Response:
		return Response(telegram_bot.get_users_as_dict())

class TelegramBotUserView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	@check_telegram_bot_user_id
	def delete(self, request: Request, telegram_bot: TelegramBot, telegram_bot_user: TelegramBotUser) -> Response:
		telegram_bot_user.delete()

		return Response({
			'message': _('Вы успешно удалили пользователя Telegram бота.'),
			'level': 'success',
		})

class TelegramBotAllowedUserView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	@check_telegram_bot_user_id
	def post(self, request: Request, telegram_bot: TelegramBot, telegram_bot_user: TelegramBotUser) -> Response:
		telegram_bot_user.is_allowed = True
		telegram_bot_user.save()

		return Response({
			'message': _('Вы успешно добавили пользователя в список разрешённых пользователей Telegram бота.'),
			'level': 'success',
		})

	@check_telegram_bot_id
	@check_telegram_bot_user_id
	def delete(self, request: Request, telegram_bot: TelegramBot, telegram_bot_user: TelegramBotUser) -> Response:
		telegram_bot_user.is_allowed = False
		telegram_bot_user.save()

		return Response({
			'message': _('Вы успешно удалили пользователя из списка разрешённых пользователей Telegram бота.'),
			'level': 'success',
		})

class TelegramBotDatabeseRecordsView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	def post(self, request: Request, telegram_bot: TelegramBot) -> Response:
		serializer = CreateTelegramBotDatabeseRecordSerializer(data=request.data)
		serializer.is_valid(raise_exception=True)

		record: dict = database_telegram_bot.insert_record(telegram_bot, serializer.validated_data['record'])

		return Response({
			'message': _('Вы успешно добавили запись в базу данных Telegram бота.'),
			'level': 'success',

			'record': record,
		})

	@check_telegram_bot_id
	def get(self, request: Request, telegram_bot: TelegramBot) -> Response:
		return Response(database_telegram_bot.get_records(telegram_bot))

class TelegramBotDatabeseRecordView(APIView):
	authentication_classes = [TokenAuthentication]
	permission_classes = [IsAuthenticated]

	@check_telegram_bot_id
	def patch(self, request: Request, telegram_bot: TelegramBot, record_id: int) -> Response:
		serializer = UpdateTelegramBotDatabeseRecordSerializer(data=request.data)
		serializer.is_valid(raise_exception=True)

		record: dict = database_telegram_bot.update_record(telegram_bot, record_id, serializer.validated_data['record'])

		return Response({
			'message': _('Вы успешно обновили запись в базе данных Telegram бота.'),
			'level': 'success',

			'record': record,
		})

	@check_telegram_bot_id
	def delete(self, request: Request, telegram_bot: TelegramBot, record_id: int) -> Response:
		database_telegram_bot.delete_record(telegram_bot, record_id)

		return Response({
			'message': _('Вы успешно удалили запись из базы данных Telegram бота.'),
			'level': 'success',
		})

constructor_telegram_bots/telegram_bot/views.py
pyguru-dev/telegrambot
from django.apps import AppConfig


class TelegramBotMenuConfig(AppConfig):
    name = 'telegram_bot_menu'

constructor_telegram_bots/telegram_bot_menu/apps.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls

from user.models import User
from telegram_bot.models import TelegramBot


class ViewsTests(TestCase):
	def setUp(self) -> None:
		self.user: User = User.objects.create(telegram_id=123456789, first_name='exg1o')
		self.telegram_bot: TelegramBot = TelegramBot.objects.create(
			owner=self.user,
			api_token='123456789:qwertyuiop',
			is_private=True
		)

	def test_telegram_bot_view(self) -> None:
		url: str = urls.reverse('telegram_bot_menu:telegram_bot', kwargs={'telegram_bot_id': self.telegram_bot.id})

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'telegram_bot.html')

	def test_telegram_bot_users_view(self) -> None:
		url: str = urls.reverse('telegram_bot_menu:telegram_bot_users', kwargs={'telegram_bot_id': self.telegram_bot.id})

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'telegram_bot_users.html')

	def test_telegram_bot_database_view(self) -> None:
		url: str = urls.reverse('telegram_bot_menu:telegram_bot_database', kwargs={'telegram_bot_id': self.telegram_bot.id})

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'telegram_bot_database.html')

	def test_telegram_bot_plugins_view(self) -> None:
		url: str = urls.reverse('telegram_bot_menu:telegram_bot_plugins', kwargs={'telegram_bot_id': self.telegram_bot.id})

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'telegram_bot_plugins.html')

	def test_telegram_bot_constructor_view(self) -> None:
		url: str = urls.reverse('telegram_bot_menu:telegram_bot_constructor', kwargs={'telegram_bot_id': self.telegram_bot.id})

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'telegram_bot_constructor/main.html')

constructor_telegram_bots/telegram_bot_menu/tests.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


app_name = 'telegram_bot_menu'
urlpatterns = [
	path('<int:telegram_bot_id>/', views.telegram_bot_view, name='telegram_bot'),
	path('<int:telegram_bot_id>/users/', views.telegram_bot_users_view, name='telegram_bot_users'),
	path('<int:telegram_bot_id>/database/', views.telegram_bot_database_view, name='telegram_bot_database'),
	path('<int:telegram_bot_id>/plugins/', views.telegram_bot_plugins_view, name='telegram_bot_plugins'),
	path('<int:telegram_bot_id>/constructor/', views.telegram_bot_constructor_view, name='telegram_bot_constructor'),
]

constructor_telegram_bots/telegram_bot_menu/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.contrib.auth.decorators import login_required
from django.shortcuts import render

from telegram_bot.decorators import check_telegram_bot_id


@login_required
@check_telegram_bot_id
def telegram_bot_view(request: HttpRequest, **context) -> HttpResponse:
	return render(request, 'telegram_bot.html', context)

@login_required
@check_telegram_bot_id
def telegram_bot_users_view(request: HttpRequest, **context) -> HttpResponse:
	return render(request, 'telegram_bot_users.html', context)

@login_required
@check_telegram_bot_id
def telegram_bot_database_view(request: HttpRequest, **context) -> HttpResponse:
	return render(request, 'telegram_bot_database.html', context)

@login_required
@check_telegram_bot_id
def telegram_bot_plugins_view(request: HttpRequest, **context) -> HttpResponse:
	return render(request, 'telegram_bot_plugins.html', context)

@login_required
@check_telegram_bot_id
def telegram_bot_constructor_view(request: HttpRequest, **context) -> HttpResponse:
	return render(request, 'telegram_bot_constructor/main.html', context)

constructor_telegram_bots/telegram_bot_menu/views.py
pyguru-dev/telegrambot
from django.contrib import admin
from django.db import models

from modeltranslation.admin import TranslationAdmin
from tinymce.widgets import TinyMCE

from .models import Update


@admin.register(Update)
class UpdateAdmin(TranslationAdmin):
	date_hierarchy = 'added_date'

	list_display = ('version', 'added_date')

	fields = ('image', 'version', 'description')
	formfield_overrides = {
		models.TextField: {
			'widget': TinyMCE,
		},
	}

constructor_telegram_bots/updates/admin.py
pyguru-dev/telegrambot
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _


class UpdatesConfig(AppConfig):
	default_auto_field = 'django.db.models.BigAutoField'

	name = 'updates'
	verbose_name = _('Обновления')

constructor_telegram_bots/updates/apps.py
pyguru-dev/telegrambot
from .models import Update


def updates(*args, **kwargs) -> dict:
    return {'updates': Update.objects.all()}

constructor_telegram_bots/updates/context_processors.py
pyguru-dev/telegrambot
from django.http import JsonResponse
from django.utils.translation import gettext as _

from rest_framework.status import *

from .models import Update

from functools import wraps


def check_update_id(func):
	@wraps(func)
	def wrapper(*args, **kwargs):
		update_id: int = kwargs.pop('update_id')

		if not Update.objects.filter(id=update_id).exists():
			return JsonResponse({
				'message': _('Обновление не найдено!'),
				'level': 'danger',
			}, status=HTTP_404_NOT_FOUND)

		return func(update=Update.objects.get(id=update_id), *args, **kwargs)
	return wrapper

constructor_telegram_bots/updates/decorators.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-01 01:44

from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    initial = True

    dependencies = [
        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
    ]

    operations = [
        migrations.CreateModel(
            name='Update',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('image', models.ImageField(upload_to='static/images/updates/')),
                ('title', models.CharField(max_length=255, verbose_name='Заголовок')),
                ('title_en', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('title_uk', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('title_ru', models.CharField(max_length=255, null=True, verbose_name='Заголовок')),
                ('description', models.TextField(verbose_name='Описание')),
                ('description_en', models.TextField(null=True, verbose_name='Описание')),
                ('description_uk', models.TextField(null=True, verbose_name='Описание')),
                ('description_ru', models.TextField(null=True, verbose_name='Описание')),
                ('views', models.BigIntegerField(default=0, verbose_name='Просмотры')),
                ('date_added', models.DateTimeField(auto_now_add=True, verbose_name='Дата добавления')),
            ],
            options={
                'verbose_name': 'Обновление',
                'verbose_name_plural': 'Обновления',
                'db_table': 'update',
            },
        ),
        migrations.CreateModel(
            name='UpdateLike',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('date_added', models.DateTimeField(auto_now_add=True)),
                ('update', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='likes', to='updates.update')),
                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
            ],
            options={
                'db_table': 'update_like',
            },
        ),
        migrations.CreateModel(
            name='UpdateComment',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('text', models.TextField()),
                ('date_added', models.DateTimeField(auto_now_add=True)),
                ('update', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='comments', to='updates.update')),
                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
            ],
            options={
                'db_table': 'update_comment',
            },
        ),
    ]

constructor_telegram_bots/updates/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-01 04:57

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('updates', '0001_initial'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='updatelike',
            name='update',
        ),
        migrations.RemoveField(
            model_name='updatelike',
            name='user',
        ),
        migrations.RemoveField(
            model_name='update',
            name='views',
        ),
        migrations.DeleteModel(
            name='UpdateComment',
        ),
        migrations.DeleteModel(
            name='UpdateLike',
        ),
    ]

constructor_telegram_bots/updates/migrations/0002_remove_updatelike_update_remove_updatelike_user_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-07-06 07:11

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('updates', '0002_remove_updatelike_update_remove_updatelike_user_and_more'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='update',
            name='title_en',
        ),
        migrations.RemoveField(
            model_name='update',
            name='title_ru',
        ),
        migrations.RemoveField(
            model_name='update',
            name='title_uk',
        ),
    ]

constructor_telegram_bots/updates/migrations/0003_remove_update_title_en_remove_update_title_ru_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-28 10:40

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('updates', '0003_remove_update_title_en_remove_update_title_ru_and_more'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='update',
            options={'ordering': ['-date_added'], 'verbose_name': 'Обновление', 'verbose_name_plural': 'Обновления'},
        ),
    ]

constructor_telegram_bots/updates/migrations/0004_alter_update_options.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 09:15

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('updates', '0004_alter_update_options'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='update',
            options={'ordering': ['-added_date'], 'verbose_name': 'Обновление', 'verbose_name_plural': 'Обновления'},
        ),
        migrations.RenameField(
            model_name='update',
            old_name='date_added',
            new_name='added_date',
        ),
        migrations.RenameField(
            model_name='update',
            old_name='title',
            new_name='version',
        ),
    ]

constructor_telegram_bots/updates/migrations/0005_alter_update_options_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-03 09:15

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('updates', '0005_alter_update_options_and_more'),
    ]

    operations = [
        migrations.AlterField(
            model_name='update',
            name='added_date',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Добавлено'),
        ),
        migrations.AlterField(
            model_name='update',
            name='version',
            field=models.CharField(max_length=255, verbose_name='Версия'),
        ),
    ]

constructor_telegram_bots/updates/migrations/0006_alter_update_added_date_alter_update_version.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-20 02:13

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('updates', '0006_alter_update_added_date_alter_update_version'),
    ]

    operations = [
        migrations.AlterField(
            model_name='update',
            name='image',
            field=models.ImageField(upload_to='updates/'),
        ),
    ]

constructor_telegram_bots/updates/migrations/0007_alter_update_image.py
pyguru-dev/telegrambot
from django.db import models
from django.utils.translation import gettext_lazy as _


class Update(models.Model):
	image = models.ImageField(upload_to='updates/')
	version = models.CharField(_('Версия'), max_length=255)
	description = models.TextField(_('Описание'))
	added_date = models.DateTimeField(_('Добавлено'), auto_now_add=True)

	class Meta:
		db_table = 'update'
		ordering = ['-added_date']

		verbose_name = _('Обновление')
		verbose_name_plural = _('Обновления')

	def __str__(self) -> str:
		return self.version

constructor_telegram_bots/updates/models.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls

from .models import Update


class UpdateModelTests(TestCase):
	def setUp(self) -> None:
		self.update = Update.objects.create(version='0.0.0-beta', description='None...')

	def test_fields(self) -> None:
		self.assertEqual(str(self.update.image), '')
		self.assertEqual(self.update.version, '0.0.0-beta')
		self.assertEqual(self.update.description, 'None...')
		self.assertIsNotNone(self.update.added_date)

class ViewsTests(TestCase):
	def test_updates_view(self) -> None:
		url: str = urls.reverse('updates')
		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'updates.html')

	def test_update_view(self) -> None:
		url: str = urls.reverse('update', kwargs={'update_id': 1})

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 404)

		Update.objects.create(version='0.0.0-beta', description='None...')

		response = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'update.html')

constructor_telegram_bots/updates/tests.py
pyguru-dev/telegrambot
from modeltranslation.translator import register, TranslationOptions

from .models import Update


@register(Update)
class UpdateTranslationOptions(TranslationOptions):
    fields = ('description',)

constructor_telegram_bots/updates/translation.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


urlpatterns = [
	path('', views.updates_view, name='updates'),
    path('<int:update_id>/', views.update_view, name='update'),
]

constructor_telegram_bots/updates/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render

from .decorators import check_update_id


def updates_view(request: HttpRequest) -> HttpResponse:
	return render(request, 'updates.html')

@check_update_id
def update_view(request: HttpRequest, **context) -> HttpResponse:
	return render(request, 'update.html', context)

constructor_telegram_bots/updates/views.py
pyguru-dev/telegrambot
from django.contrib import admin
from django.utils.translation import gettext_lazy as _

from .models import User


@admin.register(User)
class UserAdmin(admin.ModelAdmin):
	search_fields = ('telegram_id', 'first_name')
	date_hierarchy = 'joined_date'
	list_filter = ('is_staff', 'last_login', 'joined_date')

	list_display = ('id', 'telegram_id', 'first_name', 'telegram_bots_count', 'is_staff', 'last_login', 'joined_date')

	fields = ('id', 'telegram_id', 'first_name', 'telegram_bots_count', 'is_staff', 'groups', 'last_login', 'joined_date')
	readonly_fields = ('id', 'telegram_id', 'first_name', 'telegram_bots_count', 'last_login', 'joined_date')

	@admin.display(description=_('Количество Telegram ботов'))
	def telegram_bots_count(self, user: User) -> int:
		return user.telegram_bots.count()

	def has_add_permission(self, *args, **kwargs) -> bool:
		return False

constructor_telegram_bots/user/admin.py
pyguru-dev/telegrambot
from .models import User


def users(*args, **kwargs) -> dict:
    return {'users': User.objects.all()}

constructor_telegram_bots/user/context_processors.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-05-17 15:42

from django.db import migrations, models


class Migration(migrations.Migration):

    initial = True

    dependencies = [
        ('auth', '0012_alter_user_first_name_max_length'),
        ('telegram_bot', '0001_initial'),
    ]

    operations = [
        migrations.CreateModel(
            name='User',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
                ('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')),
                ('confirm_code', models.CharField(max_length=25, null=True)),
                ('is_staff', models.BooleanField(default=False)),
                ('date_joined', models.DateTimeField(auto_now_add=True)),
                ('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.group', verbose_name='groups')),
                ('telegram_bots', models.ManyToManyField(related_name='telegram_bots', to='telegram_bot.telegrambot')),
                ('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.permission', verbose_name='user permissions')),
            ],
            options={
                'db_table': 'user',
            },
        ),
    ]

constructor_telegram_bots/user/migrations/0001_initial.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-01 21:44

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0002_user_password_user_username'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='user',
            name='is_staff',
        ),
        migrations.RemoveField(
            model_name='user',
            name='telegram_bots',
        ),
        migrations.AlterField(
            model_name='user',
            name='confirm_code',
            field=models.CharField(max_length=25, null=True, unique=True),
        ),
    ]

constructor_telegram_bots/user/migrations/0003_remove_user_is_staff_remove_user_telegram_bots_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-24 13:18

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0003_remove_user_is_staff_remove_user_telegram_bots_and_more'),
    ]

    operations = [
        migrations.AlterModelOptions(
            name='user',
            options={'verbose_name': 'Пользователя', 'verbose_name_plural': 'Пользователи'},
        ),
        migrations.AddField(
            model_name='user',
            name='is_staff',
            field=models.BooleanField(default=False),
        ),
        migrations.AlterField(
            model_name='user',
            name='date_joined',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Дата присоединения'),
        ),
        migrations.AlterField(
            model_name='user',
            name='username',
            field=models.CharField(max_length=32, null=True, unique=True, verbose_name='Имя пользователя'),
        ),
    ]

constructor_telegram_bots/user/migrations/0004_alter_user_options_user_is_staff_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-24 15:18

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0004_alter_user_options_user_is_staff_and_more'),
    ]

    operations = [
        migrations.AlterField(
            model_name='user',
            name='is_staff',
            field=models.BooleanField(default=False, verbose_name='Сотрудник'),
        ),
    ]

constructor_telegram_bots/user/migrations/0005_alter_user_is_staff.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-25 22:32

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0005_alter_user_is_staff'),
    ]

    operations = [
        migrations.AlterField(
            model_name='user',
            name='password',
            field=models.CharField(max_length=25, null=True, verbose_name='Пароль'),
        ),
    ]

constructor_telegram_bots/user/migrations/0006_alter_user_password.py
pyguru-dev/telegrambot
# Generated by Django 4.2.1 on 2023-06-27 08:09

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0006_alter_user_password'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='user',
            name='password',
        ),
    ]

constructor_telegram_bots/user/migrations/0007_remove_user_password.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-11 21:28

from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    dependencies = [
        ('telegram_bot', '0014_alter_telegrambotuser_options_and_more'),
        ('user', '0007_remove_user_password'),
    ]

    operations = [
        migrations.CreateModel(
            name='UserPlugin',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('name', models.CharField(max_length=255, verbose_name='Название')),
                ('code', models.TextField(verbose_name='Код')),
                ('is_checked', models.BooleanField(verbose_name='Проверен')),
                ('date_added', models.DateTimeField(auto_now_add=True, verbose_name='Дата добавления')),
                ('telegram_bot', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='telegram_bot.telegrambot', verbose_name='Telegram бот')),
            ],
            options={
                'verbose_name': 'Плагин пользователя',
                'verbose_name_plural': 'Плагины пользователей',
                'db_table': 'user_plugin',
            },
        ),
        migrations.AddField(
            model_name='user',
            name='telegram_id',
            field=models.BigIntegerField(null=True, unique=True, verbose_name='Telegram ID'),
        ),
        migrations.CreateModel(
            name='UserPluginLog',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('message', models.TextField(verbose_name='Сообщение')),
                ('level', models.CharField(choices=[('info', 'Info'), ('success', 'Success'), ('danger', 'Danger')], default='info', max_length=7, verbose_name='Уровень')),
                ('date_added', models.DateTimeField(auto_now_add=True, verbose_name='Дата добавления')),
                ('plugin', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='user.userplugin', verbose_name='Плагин')),
                ('telegram_bot', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='telegram_bot.telegrambot', verbose_name='Telegram бот')),
                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL, verbose_name='Пользователь')),
            ],
            options={
                'verbose_name': 'Журнал плагина пользователя',
                'verbose_name_plural': 'Журналы плагинов пользователей',
                'db_table': 'user_plugin_log',
            },
        ),
        migrations.AddField(
            model_name='userplugin',
            name='user',
            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL, verbose_name='Пользователь'),
        ),
    ]

constructor_telegram_bots/user/migrations/0008_userplugin_user_telegram_id_userpluginlog_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-11 23:34

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0008_userplugin_user_telegram_id_userpluginlog_and_more'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='userpluginlog',
            name='plugin',
        ),
        migrations.RemoveField(
            model_name='userpluginlog',
            name='telegram_bot',
        ),
        migrations.RemoveField(
            model_name='userpluginlog',
            name='user',
        ),
        migrations.DeleteModel(
            name='UserPlugin',
        ),
        migrations.DeleteModel(
            name='UserPluginLog',
        ),
    ]

constructor_telegram_bots/user/migrations/0009_remove_userpluginlog_plugin_and_more.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-07-19 18:51

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0009_remove_userpluginlog_plugin_and_more'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='user',
            name='username',
        ),
        migrations.AddField(
            model_name='user',
            name='first_name',
            field=models.CharField(max_length=64, null=True, verbose_name='Имя пользователя'),
        ),
    ]

constructor_telegram_bots/user/migrations/0010_remove_user_username_user_first_name.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-12 19:16

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0010_remove_user_username_user_first_name'),
    ]

    operations = [
        migrations.RenameField(
            model_name='user',
            old_name='date_joined',
            new_name='joined_date',
        ),
    ]

constructor_telegram_bots/user/migrations/0011_rename_date_joined_user_joined_date.py
pyguru-dev/telegrambot
# Generated by Django 4.2.3 on 2023-09-12 19:17

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('user', '0011_rename_date_joined_user_joined_date'),
    ]

    operations = [
        migrations.AlterField(
            model_name='user',
            name='joined_date',
            field=models.DateTimeField(auto_now_add=True, verbose_name='Присоединился'),
        ),
    ]

constructor_telegram_bots/user/migrations/0012_alter_user_joined_date.py
pyguru-dev/telegrambot
from django.contrib.auth.base_user import BaseUserManager
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
from django.db import models
from django.utils.translation import gettext_lazy as _
from django.conf import settings
from django import urls

from rest_framework.authtoken.models import Token

from constructor_telegram_bots.functions import generate_random_string
from constructor_telegram_bots import environment

from asgiref.sync import sync_to_async
import requests


class UserManager(BaseUserManager):
	def create(self, telegram_id: int, first_name: str, **extra_fields) -> 'User':
		user: 'User' = super().create(telegram_id=telegram_id, first_name=first_name, **extra_fields)
		Token.objects.create(user=user)
		environment.create_user(user=user)
		return user

	def create_superuser(self, **fields) -> None:
		raise SyntaxError('Not support to create superuser!')

class User(AbstractBaseUser, PermissionsMixin):
	telegram_id = models.BigIntegerField('Telegram ID', unique=True, null=True)
	first_name = models.CharField(_('Имя пользователя'), max_length=64, null=True)
	password = None
	is_staff = models.BooleanField(_('Сотрудник'), default=False)
	confirm_code = models.CharField(max_length=25, unique=True, null=True)
	joined_date = models.DateTimeField(_('Присоединился'), auto_now_add=True)

	USERNAME_FIELD = 'telegram_id'

	objects = UserManager()

	class Meta:
		db_table = 'user'

		verbose_name = _('Пользователя')
		verbose_name_plural = _('Пользователи')

	def generate_login_url(self) -> str:
		if not self.confirm_code:
			self.confirm_code = generate_random_string(length=25, chars='abcdefghijklnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890')
			self.save()
		return f"{settings.SITE_DOMAIN}{urls.reverse('user:login', kwargs={'user_id': self.id, 'confirm_code': self.confirm_code})}"

	@property
	def login_url(self) -> str:
		return self.generate_login_url()

	@property
	async def alogin_url(self) -> str:
		return await sync_to_async(self.generate_login_url)()

	def update_first_name(self) -> None:
		response: requests.Response = requests.get(f'https://api.telegram.org/bot{settings.CONSTRUCTOR_TELEGRAM_BOT_API_TOKEN}/getChat?chat_id={self.telegram_id}')

		if response.status_code == 200:
			first_name: str = response.json()['result']['first_name']

			if self.first_name != first_name:
				self.first_name = first_name
				self.save()

	def get_telegram_bots_as_dict(self) -> list:
		return [telegram_bot.to_dict() for telegram_bot in self.telegram_bots.all()]

	def delete(self) -> None:
		environment.delete_user(self)
		super().delete()

	def __str__(self) -> str:
		return self.first_name

constructor_telegram_bots/user/models.py
pyguru-dev/telegrambot
from celery import shared_task

from .models import User

import time


@shared_task
def update_users_first_name():
	for user in User.objects.all():
		user.update_first_name()
		time.sleep(0.1)

constructor_telegram_bots/user/tasks.py
pyguru-dev/telegrambot
from django.test import TestCase
from django.http import HttpResponse
from django import urls
from django.conf import settings

from .models import User


class BaseTestCase(TestCase):
	def setUp(self) -> None:
		self.user: User = User.objects.create(telegram_id=123456789, first_name='exg1o')

class UserModelTests(BaseTestCase):
	def test_fields(self) -> None:
		self.assertEqual(self.user.telegram_id, 123456789)
		self.assertEqual(self.user.first_name, 'exg1o')
		self.assertFalse(self.user.is_superuser)
		self.assertFalse(self.user.is_staff)
		self.assertIsNone(self.user.confirm_code)
		self.assertIsNone(self.user.last_login)

	def test_login_url(self) -> None:
		self.assertEqual(self.user.login_url, f"{settings.SITE_DOMAIN}{urls.reverse('user:login', kwargs={'user_id': self.user.id, 'confirm_code': self.user.confirm_code})}")
		self.assertIsNotNone(self.user.confirm_code)
		self.assertIsNone(self.user.last_login)

		self.client.get(self.user.login_url)

		self.user.refresh_from_db()
		self.assertIsNone(self.user.confirm_code)
		self.assertIsNotNone(self.user.last_login)

class ViewsTests(BaseTestCase):
	def test_user_login_view(self) -> None:
		login_urls = {
			urls.reverse('user:login', kwargs={
				'user_id': 0,
				'confirm_code': 1,
			}): 'Не удалось найти пользователя!',
			urls.reverse('user:login', kwargs={
				'user_id': 1,
				'confirm_code': 0,
			}): 'Неверный код подтверждения!',
		}

		for login_url in login_urls:
			response: HttpResponse = self.client.get(login_url)
			self.assertEqual(response.status_code, 200)
			self.assertTemplateUsed(response, 'base_success_or_error.html')
			self.assertContains(response, login_urls[login_url])

		response: HttpResponse = self.client.get(self.user.login_url)
		self.assertEqual(response.status_code, 302)

	def test_user_logout_view(self) -> None:
		url: str = urls.reverse('user:logout')

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 302)

		self.client.get(self.user.login_url)

		response: HttpResponse = self.client.get(url)
		self.assertEqual(response.status_code, 200)
		self.assertTemplateUsed(response, 'base_success_or_error.html')

constructor_telegram_bots/user/tests.py
pyguru-dev/telegrambot
from django.urls import path

from . import views


app_name = 'user'
urlpatterns = [
	path('login/<int:user_id>/<str:confirm_code>/', views.user_login_view, name='login'),
	path('logout/', views.user_logout_view, name='logout'),
]

constructor_telegram_bots/user/urls.py
pyguru-dev/telegrambot
from django.http import HttpRequest, HttpResponse
from django.utils.translation import gettext as _
from django import urls
from django.shortcuts import render, redirect
from django.contrib.auth import login, logout
from django.contrib.auth.decorators import login_required

from .models import User


def user_login_view(request: HttpRequest, user_id: int, confirm_code: str) -> HttpResponse:
	context = {
		'title': _('Ошибка авторизации'),
		'meta': {'refresh': {'url': urls.reverse('home')}},
		'content': {'text': _('Автоматический переход на главную страницу через 3 секунды.')},
	}

	if not User.objects.filter(id=user_id).exists():
		context['content']['heading'] = _('Не удалось найти пользователя!')
		return render(request, 'base_success_or_error.html', context)

	user: User = User.objects.get(id=user_id)

	if user.confirm_code != confirm_code:
		context['content']['heading'] = _('Неверный код подтверждения!')
		return render(request, 'base_success_or_error.html', context)

	user.confirm_code = None
	user.save()

	login(request, user)

	return redirect(urls.reverse('personal_cabinet'))

@login_required
def user_logout_view(request: HttpRequest) -> HttpResponse:
	logout(request)

	return render(request, 'base_success_or_error.html', {
		'title': _('Выход из аккаунта'),
		'meta': {'refresh': {'url': urls.reverse('home')}},
		'content': {
			'heading': _('Успешный выход из аккаунта'),
			'text': _('Автоматический переход на главную страницу через 3 секунды.'),
		},
	})

constructor_telegram_bots/user/views.py
pyguru-dev/telegrambot
import random


random_string = ''

for num in range(50):
	random_string += random.choice('abcdefghijklnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890!@#$%^&*()_')

print(f'django-insecure-{random_string}')

scripts/generate_django_secret_key.py
pyguru-dev/telegrambot