# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\manage.py
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys


def main():
    """Run administrative tasks."""
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'dtb.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()

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\README.md
# django-telegram-bot

<p align="center">
    <img src="https://user-images.githubusercontent.com/50623190/201977740-68ef4044-9cfa-45da-8897-2a90ecfa33ae.png" align="center" height="350px" weight="350px">
</p>

Sexy Django + python-telegram-bot + Celery + Redis + Postgres + Dokku + GitHub Actions template. Production-ready Telegram bot with database, admin panel and a bunch of useful built-in methods.


⭐ graph: 
[![Sparkline](https://stars.medv.io/ohld/django-telegram-bot.svg)](https://stars.medv.io/ohld/django-telegram-bot)


### Check the example bot that uses the code from Main branch: [t.me/djangotelegrambot](https://t.me/djangotelegrambot)

## Features

* Database: Postgres, Sqlite3, MySQL - you decide!
* Admin panel (thanks to [Django](https://docs.djangoproject.com/en/3.1/intro/tutorial01/))
* Background jobs using [Celery](https://docs.celeryproject.org/en/stable/)
* [Production-ready](https://github.com/ohld/django-telegram-bot/wiki/Production-Deployment-using-Dokku) deployment using [Dokku](https://dokku.com)
* Telegram API usage in polling or [webhook mode](https://core.telegram.org/bots/api#setwebhook)
* Export all users in `.csv`
* Native telegram [commands in menu](https://github.com/ohld/django-telegram-bot/blob/main/.github/imgs/bot_commands_example.jpg)
  * In order to edit or delete these commands you'll need to use `set_my_commands` bot's method just like in [tgbot.dispatcher.setup_my_commands](https://github.com/ohld/django-telegram-bot/blob/main/tgbot/dispatcher.py#L150-L156)

Built-in Telegram bot methods:
* `/broadcast` — send message to all users (admin command)
* `/export_users` — bot sends you info about your users in .csv file (admin command)
* `/stats` — show basic bot stats 
* `/ask_for_location` — log user location when received and reverse geocode it to get country, city, etc.


## Content

* [How to run locally](https://github.com/ohld/django-telegram-bot/#how-to-run)
   * [Quickstart with polling and SQLite](https://github.com/ohld/django-telegram-bot/#quickstart-polling--sqlite)
   * [Using docker-compose](https://github.com/ohld/django-telegram-bot/#run-locally-using-docker-compose)
* [Deploy to production](https://github.com/ohld/django-telegram-bot/#deploy-to-production)
   * [Using dokku](https://github.com/ohld/django-telegram-bot/#deploy-using-dokku-step-by-step)
   * [Telegram webhook](https://github.com/ohld/django-telegram-bot/#https--telegram-bot-webhook)


# How to run

## Quickstart: Polling & SQLite

The fastest way to run the bot is to run it in polling mode using SQLite database without all Celery workers for background jobs. This should be enough for quickstart:

``` bash
git clone https://github.com/ohld/django-telegram-bot
cd django-telegram-bot
```

Create virtual environment (optional)
``` bash
python3 -m venv dtb_venv
source dtb_venv/bin/activate
```

Install all requirements:
```
pip install -r requirements.txt
```

Create `.env` file in root directory and copy-paste this or just run `cp .env_example .env`,
don't forget to change telegram token:
``` bash 
DJANGO_DEBUG=True
DATABASE_URL=sqlite:///db.sqlite3
TELEGRAM_TOKEN=<PASTE YOUR TELEGRAM TOKEN HERE>
```

Run migrations to setup SQLite database:
``` bash
python manage.py migrate
```

Create superuser to get access to admin panel:
``` bash
python manage.py createsuperuser
```

Run bot in polling mode:
``` bash
python run_polling.py 
```

If you want to open Django admin panel which will be located on http://localhost:8000/tgadmin/:
``` bash
python manage.py runserver
```

## Run locally using docker-compose
If you want just to run all the things locally, you can use Docker-compose which will start all containers for you.

### Create .env file. 
You can switch to PostgreSQL just by uncommenting it's `DATABASE_URL` and commenting SQLite variable.
```bash
cp .env_example .env
```

### Docker-compose

To run all services (Django, Postgres, Redis, Celery) at once:
``` bash
docker-compose up -d --build
```

Check status of the containers.
``` bash
docker ps -a
```
It should look similar to this:
<p align="left">
    <img src="https://github.com/ohld/django-telegram-bot/raw/main/.github/imgs/containers_status.png">
</p>

Try visit <a href="http://0.0.0.0:8000/tgadmin">Django-admin panel</a>.

### Enter django shell:

``` bash
docker exec -it dtb_django bash
```

### Create superuser for Django admin panel

``` bash
python manage.py createsuperuser
```

### To see logs of the container:

``` bash
docker logs -f dtb_django
```


# Deploy to Production 

Production stack will include these technologies:

1) Postgres as main database for Django
2) Celery + Redis + easy scalable workers 
3) Dokku as PaaS (will build app from sources and deploy it with zero downtime)

All app's services that are going to be launched in production can be found in `Procfile` file. It includes Django webserver (Telegram event processing + admin panel) and Celery workers (background and periodic jobs).

## What is Dokku and how it works

[Dokku](https://dokku.com/) is an open-source version of Heroku. 

I really like Heroku deployment approach: 
1) you push commit to Main branch of your Repo
2) in couple minutes your new app is running
3) if something breaks during deployment - old app will not be shut down

You can achieve the same approach with Dokku + Github Actions (just to trigger deployment).

Dokku uses [buildpacks](https://buildpacks.io/) technology to create a Docker image from the code. No Dockerfile needed. Speaking about Python, it requires `requirements.txt`, `Procfile` files to run the things up. Also files `DOKKU_SCALE` and `runtime.txt` are useful to tweak configs to make the deployed app even better. E.g. in `DOKKU_SCALE` you can specify how many app instances should be run behind built-in load balancer.

One disadvantage of Dokku that you should be warned about is that it can work with one server only. You can't just scale your app up to 2 machines using only small config change. You still can use several servers by providing correct .env URLs to deployed apps (e.g. DATABASE_URL) but it will require more time to setup.

## Deploy using Dokku: step-by-step

I assume that you already have [Dokku installed](https://dokku.com/docs/getting-started/installation/) on your server. Let's also assume that the address of your server is *<YOURDOMAIN.COM>* (you will need a domain to setup HTTPs for Telegram webhook support). I'd recommend to have at least [2GB RAM and 2 CPU cores](https://m.do.co/c/260555f64021).

### Create Dokku app

``` bash
dokku apps:create dtb
```

You might need to added `.env` variables to app, e.g. to specify Telegram token:

``` bash
dokku config:set dtb TELEGRAM_TOKEN=.....
```

### Postgres and Redis

**Postgres** and **Redis** are configured as Dokku plugins on a server.
They will automatically add REDIS_URL & DATABASE_URL .env vars to the app after being linked.
You might need to install these Dokku plugins before.
[Install Postgres](https://github.com/dokku/dokku-postgres),
[install Redis](https://github.com/dokku/dokku-redis).

``` bash
dokku postgres:create dtb
dokku postgres:link dtb dtb

dokku redis:create dtb
dokku redis:link dtb dtb
```

### Deploy on commit with Github Actions

Go to file [.github/workflows/dokku.yml](https://github.com/ohld/django-telegram-bot/blob/main/.github/workflows/dokku.yml):

1. Enter your host name (address of your server), 
2. Deployed dokku app name (in our case this is `dtb`),
3. Set `SSH_PRIVATE_KEY` secret variable via GitHub repo settings. This private key should have the **root ssh access** to your server.

This will trigger Dokku's zero-downtime deployment. You would probably need to fork this repo to change file. 

After that you should see a green arrow ✅ at Github Actions tab that would mean your app is deployed successfully. If you see a red cross ❌ you can find the deployed logs in Github Actions tab and find out what went wrong.

## HTTPS & Telegram bot webhook

### Why you need to setup webhook

Basic polling approach is really handy and can speed up development of Telegram bots. But it doesn't scale. Better approach is to allow Telegram servers push events (webhook messages) to your server when something happens with your Telegram bot. You can use built-in Dokku load-balancer to parallel event processing.

### HTTPS using Letsencrypt plugin

For Telegram bot API webhook usage you'll need a **https** which can be setup using [Letsencrypt Dokku plugin](https://github.com/dokku/dokku-letsencrypt). You will need to attach a domain to your Django app before and specify a email (required by Letsencrypt) - you will receive notifications when certificates would become old. Make sure you achieved a successful deployment first (your app runs at <YOURDOMAIN.COM>, check in browser).

``` bash
dokku domains:add dtb <YOURDOMAIN.COM>
dokku config:set --global DOKKU_LETSENCRYPT_EMAIL=<YOUR@EMAIL.COM>
dokku letsencrypt:enable dtb
```

### Setup Telegram Bot API webhook URL

You need to tell Telegram servers where to send events of your Telegram bot. Just open in the browser:

```
https://api.telegram.org/bot<TELEGRAM_TOKEN>/setWebhook?url=https://<YOURDOMAIN.COM>/super_secter_webhook/
```


### After deployment

You can be sure that your app is deployed successfully if you see a green arrow at the latest workflow at Github Actions tab.

You would need to create a superuser to access an admin panel at https://<YOURDOMAIN.COM>/tgadmin. This can be done using a standard way using django shell:


### Open shell in deployed app 
``` shell
dokku enter dtb web 
```

### Create Django super user
Being inside a container:
``` bash
python manage.py createsuperuser
```

After that you can open admin panel of your deployed app which is located at https://<YOURDOMAIN.COM>/tgadmin.

### Read app logs

``` bash
dokku logs dtb -t
```


----

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\requirements.txt
pytz
requests
python-dotenv

ipython==8.5.0  # enhanced python interpreter

# Django
django==3.2.9
django-cors-headers==3.13.0
django-debug-toolbar==3.6.0
whitenoise==6.2.0  # for serving static files

# Django 3.0 async requirements
gunicorn==20.1.0
uvicorn==0.18.3

# Databases
psycopg2-binary==2.9.9
dj-database-url==1.0.0

# Distributed async tasks
celery==5.2.7
redis==4.3.4
django-celery-beat==2.3.0

# Telegram
python-telegram-bot==13.15  # last sync version

# monitoring
# sentry-sdk

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\runtime.txt
python-3.8.3
================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\run_polling.py
import os, django

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

from telegram import Bot
from telegram.ext import Updater

from dtb.settings import TELEGRAM_TOKEN
from tgbot.dispatcher import setup_dispatcher


def run_polling(tg_token: str = TELEGRAM_TOKEN):
    """ Run bot in polling mode """
    updater = Updater(tg_token, use_context=True)

    dp = updater.dispatcher
    dp = setup_dispatcher(dp)

    bot_info = Bot(tg_token).get_me()
    bot_link = f"https://t.me/{bot_info['username']}"

    print(f"Polling of '{bot_link}' has started")
    # it is really useful to send '👋' emoji to developer
    # when you run local test
    # bot.send_message(text='👋', chat_id=<YOUR TELEGRAM ID>)

    updater.start_polling()
    updater.idle()


if __name__ == "__main__":
    run_polling()

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\asgi.py
"""
ASGI config for dtb project.

It exposes the ASGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
"""

import os

from django.core.asgi import get_asgi_application

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

application = get_asgi_application()

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\celery.py
import os
from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'dtb.settings')

app = Celery('dtb')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
app.conf.enable_utc = False

    
================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\settings.py
import logging
import os
import sys

import dj_database_url
import dotenv

from pathlib import Path

# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent


# Load env variables from file
dotenv_file = BASE_DIR / ".env"
if os.path.isfile(dotenv_file):
    dotenv.load_dotenv(dotenv_file)


# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = os.getenv(
    "DJANGO_SECRET_KEY",
    'x%#3&%giwv8f0+%r946en7z&d@9*rc$sl0qoql56xr%bh^w2mj',
)

if os.environ.get('DJANGO_DEBUG', default=False) in ['True', 'true', '1', True]:
    DEBUG = True
else:
    DEBUG = False

ALLOWED_HOSTS = ["*",]  # since Telegram uses a lot of IPs for webhooks


INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    # 3rd party apps
    'django_celery_beat',
    'debug_toolbar',

    # local apps
    'users.apps.UsersConfig',
]

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

    'whitenoise.middleware.WhiteNoiseMiddleware',
    'corsheaders.middleware.CorsMiddleware',
    'debug_toolbar.middleware.DebugToolbarMiddleware',

    'django.middleware.common.CommonMiddleware',
]

INTERNAL_IPS = [
    # ...
    '127.0.0.1',
    # ...
]

DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'

CORS_ORIGIN_ALLOW_ALL = True
CORS_ALLOW_CREDENTIALS = True

ROOT_URLCONF = 'dtb.urls'

TEMPLATES = [
    {
        'BACKEND': 'django.template.backends.django.DjangoTemplates',
        'DIRS': [],
        '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',
            ],
        },
    },
]

WSGI_APPLICATION = 'dtb.wsgi.application'
ASGI_APPLICATION = 'dtb.asgi.application'


# Database
# https://docs.djangoproject.com/en/3.0/ref/settings/#databases

DATABASES = {
    'default': dj_database_url.config(conn_max_age=600, default="sqlite:///db.sqlite3"),
}

# Password validation
# https://docs.djangoproject.com/en/3.0/ref/settings/#auth-password-validators

AUTH_PASSWORD_VALIDATORS = [
    {
        'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
    },
    {
        'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
    },
    {
        'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
    },
    {
        'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
    },
]


# Internationalization
# https://docs.djangoproject.com/en/3.0/topics/i18n/

LANGUAGE_CODE = 'en-us'
TIME_ZONE = 'UTC'
USE_I18N = True
USE_L10N = True
USE_TZ = True


# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/3.0/howto/static-files/
STATICFILES_STORAGE = 'whitenoise.storage.CompressedManifestStaticFilesStorage'

PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
STATIC_URL = '/static/'
STATIC_ROOT = os.path.join(PROJECT_ROOT, 'static')


# -----> CELERY
REDIS_URL = os.getenv('REDIS_URL', 'redis://redis:6379')
BROKER_URL = REDIS_URL
CELERY_BROKER_URL = REDIS_URL
CELERY_RESULT_BACKEND = REDIS_URL
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = TIME_ZONE
CELERY_TASK_DEFAULT_QUEUE = 'default'


# -----> TELEGRAM
TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
if TELEGRAM_TOKEN is None:
    logging.error(
        "Please provide TELEGRAM_TOKEN in .env file.\n"
        "Example of .env file: https://github.com/ohld/django-telegram-bot/blob/main/.env_example"
    )
    sys.exit(1)

TELEGRAM_LOGS_CHAT_ID = os.getenv("TELEGRAM_LOGS_CHAT_ID", default=None)

# -----> SENTRY
# import sentry_sdk
# from sentry_sdk.integrations.django import DjangoIntegration
# from sentry_sdk.integrations.celery import CeleryIntegration
# from sentry_sdk.integrations.redis import RedisIntegration

# sentry_sdk.init(
#     dsn="INPUT ...ingest.sentry.io/ LINK",
#     integrations=[
#         DjangoIntegration(),
#         CeleryIntegration(),
#         RedisIntegration(),
#     ],
#     traces_sample_rate=0.1,

#     # If you wish to associate users to errors (assuming you are using
#     # django.contrib.auth) you may enable sending PII data.
#     send_default_pii=True
# )

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\urls.py
"""dtb URL Configuration

The `urlpatterns` list routes URLs to views. For more information please see:
    https://docs.djangoproject.com/en/3.1/topics/http/urls/
Examples:
Function views
    1. Add an import:  from my_app import views
    2. Add a URL to urlpatterns:  path('', views.home, name='home')
Class-based views
    1. Add an import:  from other_app.views import Home
    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
Including another URLconf
    1. Import the include() function: from django.urls import include, path
    2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
"""
import debug_toolbar
from django.contrib import admin
from django.urls import path, include
from django.views.decorators.csrf import csrf_exempt

from . import views

urlpatterns = [
    path('tgadmin/', admin.site.urls),
    path('__debug__/', include(debug_toolbar.urls)),
    path('', views.index, name="index"),
    path('super_secter_webhook/', csrf_exempt(views.TelegramBotWebhookView.as_view())),
]

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\views.py
import json
import logging
from django.views import View
from django.http import JsonResponse
from telegram import Update

from dtb.celery import app
from dtb.settings import DEBUG
from tgbot.dispatcher import dispatcher
from tgbot.main import bot

logger = logging.getLogger(__name__)


@app.task(ignore_result=True)
def process_telegram_event(update_json):
    update = Update.de_json(update_json, bot)
    dispatcher.process_update(update)


def index(request):
    return JsonResponse({"error": "sup hacker"})


class TelegramBotWebhookView(View):
    # WARNING: if fail - Telegram webhook will be delivered again.
    # Can be fixed with async celery task execution
    def post(self, request, *args, **kwargs):
        if DEBUG:
            process_telegram_event(json.loads(request.body))
        else:
            # Process Telegram event in Celery worker (async)
            # Don't forget to run it and & Redis (message broker for Celery)!
            # Locally, You can run all of these services via docker-compose.yml
            process_telegram_event.delay(json.loads(request.body))

        # e.g. remove buttons, typing event
        return JsonResponse({"ok": "POST request processed"})

    def get(self, request, *args, **kwargs):  # for debug
        return JsonResponse({"ok": "Get request received! But nothing done"})

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\wsgi.py
"""
WSGI config for dtb project.

It exposes the WSGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/wsgi/
"""

import os

from django.core.wsgi import get_wsgi_application

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

application = get_wsgi_application()

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\dtb\__init__.py
from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)
================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\dispatcher.py
"""
    Telegram event handlers
"""
from telegram.ext import (
    Dispatcher, Filters,
    CommandHandler, MessageHandler,
    CallbackQueryHandler,
)

from dtb.settings import DEBUG
from tgbot.handlers.broadcast_message.manage_data import CONFIRM_DECLINE_BROADCAST
from tgbot.handlers.broadcast_message.static_text import broadcast_command
from tgbot.handlers.onboarding.manage_data import SECRET_LEVEL_BUTTON

from tgbot.handlers.utils import files, error
from tgbot.handlers.admin import handlers as admin_handlers
from tgbot.handlers.location import handlers as location_handlers
from tgbot.handlers.onboarding import handlers as onboarding_handlers
from tgbot.handlers.broadcast_message import handlers as broadcast_handlers
from tgbot.main import bot


def setup_dispatcher(dp):
    """
    Adding handlers for events from Telegram
    """
    # onboarding
    dp.add_handler(CommandHandler("start", onboarding_handlers.command_start))

    # admin commands
    dp.add_handler(CommandHandler("admin", admin_handlers.admin))
    dp.add_handler(CommandHandler("stats", admin_handlers.stats))
    dp.add_handler(CommandHandler('export_users', admin_handlers.export_users))

    # location
    dp.add_handler(CommandHandler("ask_location", location_handlers.ask_for_location))
    dp.add_handler(MessageHandler(Filters.location, location_handlers.location_handler))

    # secret level
    dp.add_handler(CallbackQueryHandler(onboarding_handlers.secret_level, pattern=f"^{SECRET_LEVEL_BUTTON}"))

    # broadcast message
    dp.add_handler(
        MessageHandler(Filters.regex(rf'^{broadcast_command}(/s)?.*'), broadcast_handlers.broadcast_command_with_message)
    )
    dp.add_handler(
        CallbackQueryHandler(broadcast_handlers.broadcast_decision_handler, pattern=f"^{CONFIRM_DECLINE_BROADCAST}")
    )

    # files
    dp.add_handler(MessageHandler(
        Filters.animation, files.show_file_id,
    ))

    # handling errors
    dp.add_error_handler(error.send_stacktrace_to_tg_chat)

    # EXAMPLES FOR HANDLERS
    # dp.add_handler(MessageHandler(Filters.text, <function_handler>))
    # dp.add_handler(MessageHandler(
    #     Filters.document, <function_handler>,
    # ))
    # dp.add_handler(CallbackQueryHandler(<function_handler>, pattern="^r\d+_\d+"))
    # dp.add_handler(MessageHandler(
    #     Filters.chat(chat_id=int(TELEGRAM_FILESTORAGE_ID)),
    #     # & Filters.forwarded & (Filters.photo | Filters.video | Filters.animation),
    #     <function_handler>,
    # ))

    return dp


n_workers = 0 if DEBUG else 4
dispatcher = setup_dispatcher(Dispatcher(bot, update_queue=None, workers=n_workers, use_context=True))

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\main.py
import logging
import sys

import telegram
from telegram import Bot

from dtb.settings import TELEGRAM_TOKEN


bot = Bot(TELEGRAM_TOKEN)
TELEGRAM_BOT_USERNAME = bot.get_me()["username"]
# Global variable - the best way I found to init Telegram bot
try:
    pass
except telegram.error.Unauthorized:
    logging.error("Invalid TELEGRAM_TOKEN.")
    sys.exit(1)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\system_commands.py
from typing import Dict

from telegram import Bot, BotCommand

from tgbot.main import bot


def set_up_commands(bot_instance: Bot) -> None:

    langs_with_commands: Dict[str, Dict[str, str]] = {
        'en': {
            'start': 'Start django bot 🚀',
            'stats': 'Statistics of bot 📊',
            'admin': 'Show admin info ℹ️',
            'ask_location': 'Send location 📍',
            'broadcast': 'Broadcast message 📨',
            'export_users': 'Export users.csv 👥',
        },
        'es': {
            'start': 'Iniciar el bot de django 🚀',
            'stats': 'Estadísticas de bot 📊',
            'admin': 'Mostrar información de administrador ℹ️',
            'ask_location': 'Enviar ubicación 📍',
            'broadcast': 'Mensaje de difusión 📨',
            'export_users': 'Exportar users.csv 👥',
        },
        'fr': {
            'start': 'Démarrer le bot Django 🚀',
            'stats': 'Statistiques du bot 📊',
            'admin': "Afficher les informations d'administrateur ℹ️",
            'ask_location': 'Envoyer emplacement 📍',
            'broadcast': 'Message de diffusion 📨',
            "export_users": 'Exporter users.csv 👥',
        },
        'ru': {
            'start': 'Запустить django бота 🚀',
            'stats': 'Статистика бота 📊',
            'admin': 'Показать информацию для админов ℹ️',
            'broadcast': 'Отправить сообщение 📨',
            'ask_location': 'Отправить локацию 📍',
            'export_users': 'Экспорт users.csv 👥',
        }
    }

    bot_instance.delete_my_commands()
    for language_code in langs_with_commands:
        bot_instance.set_my_commands(
            language_code=language_code,
            commands=[
                BotCommand(command, description) for command, description in langs_with_commands[language_code].items()
            ]
        )


set_up_commands(bot)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\admin\handlers.py
from datetime import timedelta

from django.utils.timezone import now
from telegram import ParseMode, Update
from telegram.ext import CallbackContext

from tgbot.handlers.admin import static_text
from tgbot.handlers.admin.utils import _get_csv_from_qs_values
from tgbot.handlers.utils.decorators import admin_only, send_typing_action
from users.models import User


@admin_only
def admin(update: Update, context: CallbackContext) -> None:
    """ Show help info about all secret admins commands """
    update.message.reply_text(static_text.secret_admin_commands)


@admin_only
def stats(update: Update, context: CallbackContext) -> None:
    """ Show help info about all secret admins commands """
    text = static_text.users_amount_stat.format(
        user_count=User.objects.count(),  # count may be ineffective if there are a lot of users.
        active_24=User.objects.filter(updated_at__gte=now() - timedelta(hours=24)).count()
    )

    update.message.reply_text(
        text,
        parse_mode=ParseMode.HTML,
        disable_web_page_preview=True,
    )


@admin_only
@send_typing_action
def export_users(update: Update, context: CallbackContext) -> None:
    # in values argument you can specify which fields should be returned in output csv
    users = User.objects.all().values()
    csv_users = _get_csv_from_qs_values(users)
    update.message.reply_document(csv_users)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\admin\static_text.py
command_start = '/stats'

secret_admin_commands = f"⚠️ Secret Admin commands\n" \
                        f"{command_start} - bot stats"

users_amount_stat = "<b>Users</b>: {user_count}\n" \
                    "<b>24h active</b>: {active_24}"

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\admin\utils.py
import io
import csv

from datetime import datetime
from django.db.models import QuerySet
from typing import Dict


def _get_csv_from_qs_values(queryset: QuerySet[Dict], filename: str = 'users'):
    keys = queryset[0].keys()

    # csv module can write data in io.StringIO buffer only
    s = io.StringIO()
    dict_writer = csv.DictWriter(s, fieldnames=keys)
    dict_writer.writeheader()
    dict_writer.writerows(queryset)
    s.seek(0)

    # python-telegram-bot library can send files only from io.BytesIO buffer
    # we need to convert StringIO to BytesIO
    buf = io.BytesIO()

    # extract csv-string, convert it to bytes and write to buffer
    buf.write(s.getvalue().encode())
    buf.seek(0)

    # set a filename with file's extension
    buf.name = f"{filename}__{datetime.now().strftime('%Y.%m.%d.%H.%M')}.csv"

    return buf

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\admin\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\broadcast_message\handlers.py
import re

import telegram
from telegram import Update
from telegram.ext import CallbackContext

from dtb.settings import DEBUG
from .manage_data import CONFIRM_DECLINE_BROADCAST, CONFIRM_BROADCAST
from .keyboards import keyboard_confirm_decline_broadcasting
from .static_text import broadcast_command, broadcast_wrong_format, broadcast_no_access, error_with_html, \
    message_is_sent, declined_message_broadcasting
from users.models import User
from users.tasks import broadcast_message


def broadcast_command_with_message(update: Update, context: CallbackContext):
    """ Type /broadcast <some_text>. Then check your message in HTML format and broadcast to users."""
    u = User.get_user(update, context)

    if not u.is_admin:
        update.message.reply_text(
            text=broadcast_no_access,
        )
    else:
        if update.message.text == broadcast_command:
            # user typed only command without text for the message.
            update.message.reply_text(
                text=broadcast_wrong_format,
                parse_mode=telegram.ParseMode.HTML,
            )
            return

        text = f"{update.message.text.replace(f'{broadcast_command} ', '')}"
        markup = keyboard_confirm_decline_broadcasting()

        try:
            update.message.reply_text(
                text=text,
                parse_mode=telegram.ParseMode.HTML,
                reply_markup=markup,
            )
        except telegram.error.BadRequest as e:
            update.message.reply_text(
                text=error_with_html.format(reason=e),
                parse_mode=telegram.ParseMode.HTML,
            )


def broadcast_decision_handler(update: Update, context: CallbackContext) -> None:
    # callback_data: CONFIRM_DECLINE_BROADCAST variable from manage_data.py
    """ Entered /broadcast <some_text>.
        Shows text in HTML style with two buttons:
        Confirm and Decline
    """
    broadcast_decision = update.callback_query.data[len(CONFIRM_DECLINE_BROADCAST):]

    entities_for_celery = update.callback_query.message.to_dict().get('entities')
    entities, text = update.callback_query.message.entities, update.callback_query.message.text

    if broadcast_decision == CONFIRM_BROADCAST:
        admin_text = message_is_sent
        user_ids = list(User.objects.all().values_list('user_id', flat=True))

        if DEBUG:
            broadcast_message(
                user_ids=user_ids,
                text=text,
                entities=entities_for_celery,
            )
        else:
            # send in async mode via celery
            broadcast_message.delay(
                user_ids=user_ids,
                text=text,
                entities=entities_for_celery,
            )
    else:
        context.bot.send_message(
            chat_id=update.callback_query.message.chat_id,
            text=declined_message_broadcasting,
        )
        admin_text = text

    context.bot.edit_message_text(
        text=admin_text,
        chat_id=update.callback_query.message.chat_id,
        message_id=update.callback_query.message.message_id,
        entities=None if broadcast_decision == CONFIRM_BROADCAST else entities,
    )

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\broadcast_message\keyboards.py
from telegram import InlineKeyboardButton, InlineKeyboardMarkup

from tgbot.handlers.broadcast_message.manage_data import CONFIRM_DECLINE_BROADCAST, CONFIRM_BROADCAST, DECLINE_BROADCAST
from tgbot.handlers.broadcast_message.static_text import confirm_broadcast, decline_broadcast


def keyboard_confirm_decline_broadcasting() -> InlineKeyboardMarkup:
    buttons = [[
        InlineKeyboardButton(confirm_broadcast, callback_data=f'{CONFIRM_DECLINE_BROADCAST}{CONFIRM_BROADCAST}'),
        InlineKeyboardButton(decline_broadcast, callback_data=f'{CONFIRM_DECLINE_BROADCAST}{DECLINE_BROADCAST}')
    ]]

    return InlineKeyboardMarkup(buttons)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\broadcast_message\manage_data.py
CONFIRM_DECLINE_BROADCAST = 'CNFM_DCLN_BRDCST'
CONFIRM_BROADCAST = 'CONFIRM'
DECLINE_BROADCAST = 'DECLINE'

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\broadcast_message\static_text.py
broadcast_command = '/broadcast'
broadcast_no_access = "Sorry, you don't have access to this function."
broadcast_wrong_format = f'To send message to all your users,' \
                         f' type {broadcast_command} command with text separated by space.\n' \
                         f'For example:\n' \
                         f'{broadcast_command} Hello, my users! This <b>bold text</b> is for you, ' \
                         f'as well as this <i>italic text.</i>\n\n' \
                         f'Examples of using <code>HTML</code> style you can found <a href="https://core.telegram.org/bots/api#html-style">here</a>.'
confirm_broadcast = "Confirm ✅"
decline_broadcast = "Decline ❌"
message_is_sent = "Message is sent ✅"
declined_message_broadcasting = "Message broadcasting is declined ❌"
error_with_html = "Can't parse your text in <code>HTML</code> style. Reason: \n{reason}"

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\broadcast_message\utils.py
from typing import Union, Optional, Dict, List

import telegram
from telegram import MessageEntity, InlineKeyboardButton, InlineKeyboardMarkup

from dtb.settings import TELEGRAM_TOKEN
from users.models import User


def from_celery_markup_to_markup(celery_markup: Optional[List[List[Dict]]]) -> Optional[InlineKeyboardMarkup]:
    markup = None
    if celery_markup:
        markup = []
        for row_of_buttons in celery_markup:
            row = []
            for button in row_of_buttons:
                row.append(
                    InlineKeyboardButton(
                        text=button['text'],
                        callback_data=button.get('callback_data'),
                        url=button.get('url'),
                    )
                )
            markup.append(row)
        markup = InlineKeyboardMarkup(markup)
    return markup


def from_celery_entities_to_entities(celery_entities: Optional[List[Dict]] = None) -> Optional[List[MessageEntity]]:
    entities = None
    if celery_entities:
        entities = [
            MessageEntity(
                type=entity['type'],
                offset=entity['offset'],
                length=entity['length'],
                url=entity.get('url'),
                language=entity.get('language'),
            )
            for entity in celery_entities
        ]
    return entities


def send_one_message(
    user_id: Union[str, int],
    text: str,
    parse_mode: Optional[str] = telegram.ParseMode.HTML,
    reply_markup: Optional[List[List[Dict]]] = None,
    reply_to_message_id: Optional[int] = None,
    disable_web_page_preview: Optional[bool] = None,
    entities: Optional[List[MessageEntity]] = None,
    tg_token: str = TELEGRAM_TOKEN,
) -> bool:
    bot = telegram.Bot(tg_token)
    try:
        m = bot.send_message(
            chat_id=user_id,
            text=text,
            parse_mode=parse_mode,
            reply_markup=reply_markup,
            reply_to_message_id=reply_to_message_id,
            disable_web_page_preview=disable_web_page_preview,
            entities=entities,
        )
    except telegram.error.Unauthorized:
        print(f"Can't send message to {user_id}. Reason: Bot was stopped.")
        User.objects.filter(user_id=user_id).update(is_blocked_bot=True)
        success = False
    else:
        success = True
        User.objects.filter(user_id=user_id).update(is_blocked_bot=False)
    return success

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\broadcast_message\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\location\handlers.py
import telegram
from telegram import Update
from telegram.ext import CallbackContext

from tgbot.handlers.location.static_text import share_location, thanks_for_location
from tgbot.handlers.location.keyboards import send_location_keyboard
from users.models import User, Location


def ask_for_location(update: Update, context: CallbackContext) -> None:
    """ Entered /ask_location command"""
    u = User.get_user(update, context)

    context.bot.send_message(
        chat_id=u.user_id,
        text=share_location,
        reply_markup=send_location_keyboard()
    )


def location_handler(update: Update, context: CallbackContext) -> None:
    # receiving user's location
    u = User.get_user(update, context)
    lat, lon = update.message.location.latitude, update.message.location.longitude
    Location.objects.create(user=u, latitude=lat, longitude=lon)

    update.message.reply_text(
        thanks_for_location,
        reply_markup=telegram.ReplyKeyboardRemove(),
    )

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\location\keyboards.py
from telegram import ReplyKeyboardMarkup, KeyboardButton

from tgbot.handlers.location.static_text import SEND_LOCATION


def send_location_keyboard() -> ReplyKeyboardMarkup:
    # resize_keyboard=False will make this button appear on half screen (become very large).
    # Likely, it will increase click conversion but may decrease UX quality.
    return ReplyKeyboardMarkup(
        [[KeyboardButton(text=SEND_LOCATION, request_location=True)]],
        resize_keyboard=True
    )

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\location\static_text.py
SEND_LOCATION = "Send 🌏🌎🌍"
share_location = "Would you mind sharing your location?"
thanks_for_location = "Thanks for 🌏🌎🌍"

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\location\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\onboarding\handlers.py
import datetime

from django.utils import timezone
from telegram import ParseMode, Update
from telegram.ext import CallbackContext

from tgbot.handlers.onboarding import static_text
from tgbot.handlers.utils.info import extract_user_data_from_update
from users.models import User
from tgbot.handlers.onboarding.keyboards import make_keyboard_for_start_command


def command_start(update: Update, context: CallbackContext) -> None:
    u, created = User.get_user_and_created(update, context)

    if created:
        text = static_text.start_created.format(first_name=u.first_name)
    else:
        text = static_text.start_not_created.format(first_name=u.first_name)

    update.message.reply_text(text=text,
                              reply_markup=make_keyboard_for_start_command())


def secret_level(update: Update, context: CallbackContext) -> None:
    # callback_data: SECRET_LEVEL_BUTTON variable from manage_data.py
    """ Pressed 'secret_level_button_text' after /start command"""
    user_id = extract_user_data_from_update(update)['user_id']
    text = static_text.unlock_secret_room.format(
        user_count=User.objects.count(),
        active_24=User.objects.filter(updated_at__gte=timezone.now() - datetime.timedelta(hours=24)).count()
    )

    context.bot.edit_message_text(
        text=text,
        chat_id=user_id,
        message_id=update.callback_query.message.message_id,
        parse_mode=ParseMode.HTML
    )
================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\onboarding\keyboards.py
from telegram import InlineKeyboardButton, InlineKeyboardMarkup

from tgbot.handlers.onboarding.manage_data import SECRET_LEVEL_BUTTON
from tgbot.handlers.onboarding.static_text import github_button_text, secret_level_button_text


def make_keyboard_for_start_command() -> InlineKeyboardMarkup:
    buttons = [[
        InlineKeyboardButton(github_button_text, url="https://github.com/ohld/django-telegram-bot"),
        InlineKeyboardButton(secret_level_button_text, callback_data=f'{SECRET_LEVEL_BUTTON}')
    ]]

    return InlineKeyboardMarkup(buttons)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\onboarding\manage_data.py
SECRET_LEVEL_BUTTON = 'SCRT_LVL'

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\onboarding\static_text.py
start_created = "Sup, {first_name}!"
start_not_created = "Welcome back, {first_name}!"
unlock_secret_room = "Congratulations! You've opened a secret room👁‍🗨. There is some information for you:\n" \
           "<b>Users</b>: {user_count}\n" \
           "<b>24h active</b>: {active_24}"
github_button_text = "GitHub"
secret_level_button_text = "Secret level🗝"

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\onboarding\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\utils\decorators.py
from functools import wraps
from typing import Callable

from telegram import Update, ChatAction
from telegram.ext import CallbackContext

from users.models import User


def admin_only(func: Callable):
    """
    Admin only decorator
    Used for handlers that only admins have access to
    """

    @wraps(func)
    def wrapper(update: Update, context: CallbackContext, *args, **kwargs):
        user = User.get_user(update, context)

        if not user.is_admin:
            return

        return func(update, context, *args, **kwargs)

    return wrapper


def send_typing_action(func: Callable):
    """Sends typing action while processing func command."""

    @wraps(func)
    def command_func(update: Update, context: CallbackContext, *args, **kwargs):
        update.effective_chat.send_chat_action(ChatAction.TYPING)
        return func(update, context, *args, **kwargs)

    return command_func

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\utils\error.py
import logging
import traceback
import html

import telegram
from telegram import Update
from telegram.ext import CallbackContext

from dtb.settings import TELEGRAM_LOGS_CHAT_ID
from users.models import User


def send_stacktrace_to_tg_chat(update: Update, context: CallbackContext) -> None:
    u = User.get_user(update, context)

    logging.error("Exception while handling an update:", exc_info=context.error)

    tb_list = traceback.format_exception(None, context.error, context.error.__traceback__)
    tb_string = ''.join(tb_list)

    # Build the message with some markup and additional information about what happened.
    # You might need to add some logic to deal with messages longer than the 4096 character limit.
    message = (
        f'An exception was raised while handling an update\n'
        f'<pre>{html.escape(tb_string)}</pre>'
    )

    user_message = """
😔 Something broke inside the bot.
It is because we are constantly improving our service but sometimes we might forget to test some basic stuff.
We already received all the details to fix the issue.
Return to /start
"""
    context.bot.send_message(
        chat_id=u.user_id,
        text=user_message,
    )

    admin_message = f"⚠️⚠️⚠️ for {u.tg_str}:\n{message}"[:4090]
    if TELEGRAM_LOGS_CHAT_ID:
        context.bot.send_message(
            chat_id=TELEGRAM_LOGS_CHAT_ID,
            text=admin_message,
            parse_mode=telegram.ParseMode.HTML,
        )
    else:
        logging.error(admin_message)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\utils\files.py
"""
'document': {
    'file_name': 'preds (4).csv', 'mime_type': 'text/csv', 
    'file_id': 'BQACAgIAAxkBAAIJ8F-QAVpXcgUgCUtr2OAHN-OC_2bmAAJwBwAC53CASIpMq-3ePqBXGwQ', 
    'file_unique_id': 'AgADcAcAAudwgEg', 'file_size': 28775
}
'photo': [
    {'file_id': 'AgACAgIAAxkBAAIJ-F-QCOHZUv6Kmf_Z3eVSmByix_IwAAOvMRvncIBIYJQP2Js-sAWGaBiVLgADAQADAgADbQADjpMFAAEbBA', 'file_unique_id': 'AQADhmgYlS4AA46TBQAB', 'file_size': 13256, 'width': 148, 'height': 320}, 
    {'file_id': 'AgACAgIAAxkBAAIJ-F-QCOHZUv6Kmf_Z3eVSmByix_IwAAOvMRvncIBIYJQP2Js-sAWGaBiVLgADAQADAgADeAADkJMFAAEbBA', 'file_unique_id': 'AQADhmgYlS4AA5CTBQAB', 'file_size': 50857, 'width': 369, 'height': 800}, 
    {'file_id': 'AgACAgIAAxkBAAIJ-F-QCOHZUv6Kmf_Z3eVSmByix_IwAAOvMRvncIBIYJQP2Js-sAWGaBiVLgADAQADAgADeQADj5MFAAEbBA', 'file_unique_id': 'AQADhmgYlS4AA4-TBQAB', 'file_size': 76018, 'width': 591, 'height': 1280}
]
'video_note': {
    'duration': 2, 'length': 300, 
    'thumb': {'file_id': 'AAMCAgADGQEAAgn_XaLgADAQAHbQADQCYAAhsE', 'file_unique_id': 'AQADWoxsmi4AA0AmAAI', 'file_size': 11684, 'width': 300, 'height': 300}, 
    'file_id': 'DQACAgIAAxkBAAIJCASO6_6Hj8qY3PGwQ', 'file_unique_id': 'AgADeQcAAudwgEg', 'file_size': 102840
}
'voice': {
    'duration': 1, 'mime_type': 'audio/ogg', 
    'file_id': 'AwACAgIAAxkBAAIKAAFfkAu_8Ntpv8n9WWHETutijg20nAACegcAAudwgEi8N3Tjeom0IxsE', 
    'file_unique_id': 'AgADegcAAudwgEg', 'file_size': 4391
}
'sticker': {
    'width': 512, 'height': 512, 'emoji': '🤔', 'set_name': 's121356145_282028_by_stickerfacebot', 'is_animated': False, 
    'thumb': {
        'file_id': 'AAMCAgADGQEAAgJUX5A5icQq_0qkwXnihR_MJuCKSRAAAmQAA3G_Owev57igO1Oj4itVTZguAAMBAAdtAAObPwACGwQ', 'file_unique_id': 'AQADK1VNmC4AA5s_AAI', 'file_size': 14242, 'width': 320, 'height': 320
    }, 
    'file_id': 'CAACAgIAAxkBAAICVF-QOYnEKv9KpMF54oUfzCbgikkQAAJkAANxvzsHr-e4oDtTo-IbBA', 'file_unique_id': 'AgADZAADcb87Bw', 'file_size': 25182
}
'video': {
    'duration': 14, 'width': 480, 'height': 854, 'mime_type': 'video/mp4', 
    'thumb': {'file_id': 'AAMCAgADGQEAAgoIX5BAQy-AfwmWLgADAQAHbQADJhAAAhsE', 'file_unique_id': 'AQAD5H8Jli4AAyYQAAI', 'file_size': 9724, 'width': 180, 'height': 320}, 
    'file_id': 'BAACAgIIAAKaCAACCcGASLV2hk3MavHGGwQ', 
    'file_unique_id': 'AgADmggAAgnBgEg', 'file_size': 1260506}, 'caption': '50603'
}
"""
from typing import Dict

import telegram
from telegram import Update
from telegram.ext import CallbackContext

from users.models import User

ALL_TG_FILE_TYPES = ["document", "video_note", "voice", "sticker", "audio", "video", "animation", "photo"]


def _get_file_id(m: Dict) -> str:
    """ extract file_id from message (and file type?) """

    for doc_type in ALL_TG_FILE_TYPES:
        if doc_type in m and doc_type != "photo":
            return m[doc_type]["file_id"]

    if "photo" in m:
        best_photo = m["photo"][-1]
        return best_photo["file_id"]


def show_file_id(update: Update, context: CallbackContext) -> None:
    """ Returns file_id of the attached file/media """
    u = User.get_user(update, context)

    if u.is_admin:
        update_json = update.to_dict()
        file_id = _get_file_id(update_json["message"])
        message_id = update_json["message"]["message_id"]
        update.message.reply_text(
            text=f"`{file_id}`",
            parse_mode=telegram.ParseMode.HTML,
            reply_to_message_id=message_id
        )

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\utils\info.py
from typing import Dict

from telegram import Update


def extract_user_data_from_update(update: Update) -> Dict:
    """ python-telegram-bot's Update instance --> User info """
    user = update.effective_user.to_dict()

    return dict(
        user_id=user["id"],
        is_blocked_bot=False,
        **{
            k: user[k]
            for k in ["username", "first_name", "last_name", "language_code"]
            if k in user and user[k] is not None
        },
    )

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\tgbot\handlers\utils\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\admin.py
from django.contrib import admin
from django.http import HttpResponseRedirect
from django.shortcuts import render

from dtb.settings import DEBUG

from users.models import Location
from users.models import User
from users.forms import BroadcastForm

from users.tasks import broadcast_message
from tgbot.handlers.broadcast_message.utils import send_one_message


@admin.register(User)
class UserAdmin(admin.ModelAdmin):
    list_display = [
        'user_id', 'username', 'first_name', 'last_name', 
        'language_code', 'deep_link',
        'created_at', 'updated_at', "is_blocked_bot",
    ]
    list_filter = ["is_blocked_bot", ]
    search_fields = ('username', 'user_id')

    actions = ['broadcast']

    def broadcast(self, request, queryset):
        """ Select users via check mark in django-admin panel, then select "Broadcast" to send message"""
        user_ids = queryset.values_list('user_id', flat=True).distinct().iterator()
        if 'apply' in request.POST:
            broadcast_message_text = request.POST["broadcast_text"]

            if DEBUG:  # for test / debug purposes - run in same thread
                for user_id in user_ids:
                    send_one_message(
                        user_id=user_id,
                        text=broadcast_message_text,
                    )
                self.message_user(request, f"Just broadcasted to {len(queryset)} users")
            else:
                broadcast_message.delay(text=broadcast_message_text, user_ids=list(user_ids))
                self.message_user(request, f"Broadcasting of {len(queryset)} messages has been started")

            return HttpResponseRedirect(request.get_full_path())
        else:
            form = BroadcastForm(initial={'_selected_action': user_ids})
            return render(
                request, "admin/broadcast_message.html", {'form': form, 'title': u'Broadcast message'}
            )


@admin.register(Location)
class LocationAdmin(admin.ModelAdmin):
    list_display = ['id', 'user_id', 'created_at']

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\apps.py
from django.apps import AppConfig


class UsersConfig(AppConfig):
    name = 'users'

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\forms.py
from django import forms


class BroadcastForm(forms.Form):
    _selected_action = forms.CharField(widget=forms.MultipleHiddenInput)
    broadcast_text = forms.CharField(widget=forms.Textarea)

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\models.py
from __future__ import annotations

from typing import Union, Optional, Tuple

from django.db import models
from django.db.models import QuerySet, Manager
from telegram import Update
from telegram.ext import CallbackContext

from tgbot.handlers.utils.info import extract_user_data_from_update
from utils.models import CreateUpdateTracker, nb, CreateTracker, GetOrNoneManager


class AdminUserManager(Manager):
    def get_queryset(self):
        return super().get_queryset().filter(is_admin=True)


class User(CreateUpdateTracker):
    user_id = models.PositiveBigIntegerField(primary_key=True)  # telegram_id
    username = models.CharField(max_length=32, **nb)
    first_name = models.CharField(max_length=256)
    last_name = models.CharField(max_length=256, **nb)
    language_code = models.CharField(max_length=8, help_text="Telegram client's lang", **nb)
    deep_link = models.CharField(max_length=64, **nb)

    is_blocked_bot = models.BooleanField(default=False)

    is_admin = models.BooleanField(default=False)

    objects = GetOrNoneManager()  # user = User.objects.get_or_none(user_id=<some_id>)
    admins = AdminUserManager()  # User.admins.all()

    def __str__(self):
        return f'@{self.username}' if self.username is not None else f'{self.user_id}'

    @classmethod
    def get_user_and_created(cls, update: Update, context: CallbackContext) -> Tuple[User, bool]:
        """ python-telegram-bot's Update, Context --> User instance """
        data = extract_user_data_from_update(update)
        u, created = cls.objects.update_or_create(user_id=data["user_id"], defaults=data)

        if created:
            # Save deep_link to User model
            if context is not None and context.args is not None and len(context.args) > 0:
                payload = context.args[0]
                if str(payload).strip() != str(data["user_id"]).strip():  # you can't invite yourself
                    u.deep_link = payload
                    u.save()

        return u, created

    @classmethod
    def get_user(cls, update: Update, context: CallbackContext) -> User:
        u, _ = cls.get_user_and_created(update, context)
        return u

    @classmethod
    def get_user_by_username_or_user_id(cls, username_or_user_id: Union[str, int]) -> Optional[User]:
        """ Search user in DB, return User or None if not found """
        username = str(username_or_user_id).replace("@", "").strip().lower()
        if username.isdigit():  # user_id
            return cls.objects.filter(user_id=int(username)).first()
        return cls.objects.filter(username__iexact=username).first()

    @property
    def invited_users(self) -> QuerySet[User]:
        return User.objects.filter(deep_link=str(self.user_id), created_at__gt=self.created_at)

    @property
    def tg_str(self) -> str:
        if self.username:
            return f'@{self.username}'
        return f"{self.first_name} {self.last_name}" if self.last_name else f"{self.first_name}"


class Location(CreateTracker):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    latitude = models.FloatField()
    longitude = models.FloatField()

    objects = GetOrNoneManager()

    def __str__(self):
        return f"user: {self.user}, created at {self.created_at.strftime('(%H:%M, %d %B %Y)')}"

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\tasks.py
"""
    Celery tasks. Some of them will be launched periodically from admin panel via django-celery-beat
"""

import time
from typing import Union, List, Optional, Dict

import telegram

from dtb.celery import app
from celery.utils.log import get_task_logger
from tgbot.handlers.broadcast_message.utils import send_one_message, from_celery_entities_to_entities, \
    from_celery_markup_to_markup

logger = get_task_logger(__name__)


@app.task(ignore_result=True)
def broadcast_message(
    user_ids: List[Union[str, int]],
    text: str,
    entities: Optional[List[Dict]] = None,
    reply_markup: Optional[List[List[Dict]]] = None,
    sleep_between: float = 0.4,
    parse_mode=telegram.ParseMode.HTML,
) -> None:
    """ It's used to broadcast message to big amount of users """
    logger.info(f"Going to send message: '{text}' to {len(user_ids)} users")

    entities_ = from_celery_entities_to_entities(entities)
    reply_markup_ = from_celery_markup_to_markup(reply_markup)
    for user_id in user_ids:
        try:
            send_one_message(
                user_id=user_id,
                text=text,
                entities=entities_,
                parse_mode=parse_mode,
                reply_markup=reply_markup_,
            )
            logger.info(f"Broadcast message was sent to {user_id}")
        except Exception as e:
            logger.error(f"Failed to send message to {user_id}, reason: {e}")
        time.sleep(max(sleep_between, 0.1))

    logger.info("Broadcast finished!")



================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\migrations\0001_initial.py
# Generated by Django 3.2.6 on 2021-08-25 10:43

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


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='User',
            fields=[
                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
                ('updated_at', models.DateTimeField(auto_now=True)),
                ('user_id', models.IntegerField(primary_key=True, serialize=False)),
                ('username', models.CharField(blank=True, max_length=32, null=True)),
                ('first_name', models.CharField(max_length=256)),
                ('last_name', models.CharField(blank=True, max_length=256, null=True)),
                ('language_code', models.CharField(blank=True, help_text="Telegram client's lang", max_length=8, null=True)),
                ('deep_link', models.CharField(blank=True, max_length=64, null=True)),
                ('is_blocked_bot', models.BooleanField(default=False)),
                ('is_banned', models.BooleanField(default=False)),
                ('is_admin', models.BooleanField(default=False)),
                ('is_moderator', models.BooleanField(default=False)),
            ],
            options={
                'ordering': ('-created_at',),
                'abstract': False,
            },
        ),
        migrations.CreateModel(
            name='Location',
            fields=[
                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
                ('latitude', models.FloatField()),
                ('longitude', models.FloatField()),
                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='users.user')),
            ],
            options={
                'ordering': ('-created_at',),
                'abstract': False,
            },
        ),
    ]

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\migrations\0002_alter_user_user_id.py
# Generated by Django 3.2.8 on 2021-12-01 16:54

from django.db import migrations, models


class Migration(migrations.Migration):

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

    operations = [
        migrations.AlterField(
            model_name='user',
            name='user_id',
            field=models.PositiveBigIntegerField(primary_key=True, serialize=False),
        ),
    ]

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\migrations\0003_rm_unused_fields.py
# Generated by Django 3.2.6 on 2021-12-01 17:47

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('users', '0002_alter_user_user_id'),
    ]

    operations = [
        migrations.RemoveField(
            model_name='user',
            name='is_banned',
        ),
        migrations.RemoveField(
            model_name='user',
            name='is_moderator',
        ),
    ]

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\migrations\__init__.py

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\users\templates\admin\broadcast_message.html
{% extends "admin/base_site.html" %}

{% block content %}
<form action="" method="post">{% csrf_token %}
    {{ form }}
    <input type="hidden" name="action" value="broadcast" />
    <input type="submit" name="apply" value="Send" />
</form>
{% endblock %}


================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\utils\models.py
from django.core.exceptions import ObjectDoesNotExist
from django.db import models


nb = dict(null=True, blank=True)


class CreateTracker(models.Model):
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)

    class Meta:
        abstract = True
        ordering = ('-created_at',)


class CreateUpdateTracker(CreateTracker):
    updated_at = models.DateTimeField(auto_now=True)

    class Meta(CreateTracker.Meta):
        abstract = True


class GetOrNoneManager(models.Manager):
    """returns none if object doesn't exist else model instance"""
    def get_or_none(self, **kwargs):
        try:
            return self.get(**kwargs)
        except ObjectDoesNotExist:
            return None

================================================================================

# File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\utils\__init__.py

================================================================================

# Media File Paths
# Folder: C:\Users\Shakeel\Desktop\django-telegram-bot-main\.github\imgs
File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\.github\imgs\bot_commands_example.jpg
File: C:\Users\Shakeel\Desktop\django-telegram-bot-main\.github\imgs\containers_status.png

================================================================================